1 #include "redis.h" 2 3 /*----------------------------------------------------------------------------- 4 * List API 5 *----------------------------------------------------------------------------*/ 6 7 /* Check the argument length to see if it requires us to convert the ziplist 8 * to a real list. Only check raw-encoded objects because integer encoded 9 * objects are never too long. */ 10 void listTypeTryConversion(robj *subject, robj *value) { 11 if (subject->encoding != REDIS_ENCODING_ZIPLIST) return; 12 if (value->encoding == REDIS_ENCODING_RAW && 13 sdslen(value->ptr) > server.list_max_ziplist_value) 14 listTypeConvert(subject,REDIS_ENCODING_LINKEDLIST); 15 } 16 17 void listTypePush(robj *subject, robj *value, int where) { 18 /* Check if we need to convert the ziplist */ 19 listTypeTryConversion(subject,value); 20 if (subject->encoding == REDIS_ENCODING_ZIPLIST && 21 ziplistLen(subject->ptr) >= server.list_max_ziplist_entries) 22 listTypeConvert(subject,REDIS_ENCODING_LINKEDLIST); 23 24 if (subject->encoding == REDIS_ENCODING_ZIPLIST) { 25 int pos = (where == REDIS_HEAD) ? ZIPLIST_HEAD : ZIPLIST_TAIL; 26 value = getDecodedObject(value); 27 subject->ptr = ziplistPush(subject->ptr,value->ptr,sdslen(value->ptr),pos); 28 decrRefCount(value); 29 } else if (subject->encoding == REDIS_ENCODING_LINKEDLIST) { 30 if (where == REDIS_HEAD) { 31 listAddNodeHead(subject->ptr,value); 32 } else { 33 listAddNodeTail(subject->ptr,value); 34 } 35 incrRefCount(value); 36 } else { 37 redisPanic("Unknown list encoding"); 38 } 39 } 40 41 robj *listTypePop(robj *subject, int where) { 42 robj *value = NULL; 43 if (subject->encoding == REDIS_ENCODING_ZIPLIST) { 44 unsigned char *p; 45 unsigned char *vstr; 46 unsigned int vlen; 47 long long vlong; 48 int pos = (where == REDIS_HEAD) ? 0 : -1; 49 p = ziplistIndex(subject->ptr,pos); 50 if (ziplistGet(p,&vstr,&vlen,&vlong)) { 51 if (vstr) { 52 value = createStringObject((char*)vstr,vlen); 53 } else { 54 value = createStringObjectFromLongLong(vlong); 55 } 56 /* We only need to delete an element when it exists */ 57 subject->ptr = ziplistDelete(subject->ptr,&p); 58 } 59 } else if (subject->encoding == REDIS_ENCODING_LINKEDLIST) { 60 list *list = subject->ptr; 61 listNode *ln; 62 if (where == REDIS_HEAD) { 63 ln = listFirst(list); 64 } else { 65 ln = listLast(list); 66 } 67 if (ln != NULL) { 68 value = listNodeValue(ln); 69 incrRefCount(value); 70 listDelNode(list,ln); 71 } 72 } else { 73 redisPanic("Unknown list encoding"); 74 } 75 return value; 76 } 77 78 unsigned long listTypeLength(robj *subject) { 79 if (subject->encoding == REDIS_ENCODING_ZIPLIST) { 80 return ziplistLen(subject->ptr); 81 } else if (subject->encoding == REDIS_ENCODING_LINKEDLIST) { 82 return listLength((list*)subject->ptr); 83 } else { 84 redisPanic("Unknown list encoding"); 85 } 86 } 87 88 /* Initialize an iterator at the specified index. */ 89 listTypeIterator *listTypeInitIterator(robj *subject, int index, unsigned char direction) { 90 listTypeIterator *li = zmalloc(sizeof(listTypeIterator)); 91 li->subject = subject; 92 li->encoding = subject->encoding; 93 li->direction = direction; 94 if (li->encoding == REDIS_ENCODING_ZIPLIST) { 95 li->zi = ziplistIndex(subject->ptr,index); 96 } else if (li->encoding == REDIS_ENCODING_LINKEDLIST) { 97 li->ln = listIndex(subject->ptr,index); 98 } else { 99 redisPanic("Unknown list encoding"); 100 } 101 return li; 102 } 103 104 /* Clean up the iterator. */ 105 void listTypeReleaseIterator(listTypeIterator *li) { 106 zfree(li); 107 } 108 109 /* Stores pointer to current the entry in the provided entry structure 110 * and advances the position of the iterator. Returns 1 when the current 111 * entry is in fact an entry, 0 otherwise. */ 112 int listTypeNext(listTypeIterator *li, listTypeEntry *entry) { 113 /* Protect from converting when iterating */ 114 redisAssert(li->subject->encoding == li->encoding); 115 116 entry->li = li; 117 if (li->encoding == REDIS_ENCODING_ZIPLIST) { 118 entry->zi = li->zi; 119 if (entry->zi != NULL) { 120 if (li->direction == REDIS_TAIL) 121 li->zi = ziplistNext(li->subject->ptr,li->zi); 122 else 123 li->zi = ziplistPrev(li->subject->ptr,li->zi); 124 return 1; 125 } 126 } else if (li->encoding == REDIS_ENCODING_LINKEDLIST) { 127 entry->ln = li->ln; 128 if (entry->ln != NULL) { 129 if (li->direction == REDIS_TAIL) 130 li->ln = li->ln->next; 131 else 132 li->ln = li->ln->prev; 133 return 1; 134 } 135 } else { 136 redisPanic("Unknown list encoding"); 137 } 138 return 0; 139 } 140 141 /* Return entry or NULL at the current position of the iterator. */ 142 robj *listTypeGet(listTypeEntry *entry) { 143 listTypeIterator *li = entry->li; 144 robj *value = NULL; 145 if (li->encoding == REDIS_ENCODING_ZIPLIST) { 146 unsigned char *vstr; 147 unsigned int vlen; 148 long long vlong; 149 redisAssert(entry->zi != NULL); 150 if (ziplistGet(entry->zi,&vstr,&vlen,&vlong)) { 151 if (vstr) { 152 value = createStringObject((char*)vstr,vlen); 153 } else { 154 value = createStringObjectFromLongLong(vlong); 155 } 156 } 157 } else if (li->encoding == REDIS_ENCODING_LINKEDLIST) { 158 redisAssert(entry->ln != NULL); 159 value = listNodeValue(entry->ln); 160 incrRefCount(value); 161 } else { 162 redisPanic("Unknown list encoding"); 163 } 164 return value; 165 } 166 167 void listTypeInsert(listTypeEntry *entry, robj *value, int where) { 168 robj *subject = entry->li->subject; 169 if (entry->li->encoding == REDIS_ENCODING_ZIPLIST) { 170 value = getDecodedObject(value); 171 if (where == REDIS_TAIL) { 172 unsigned char *next = ziplistNext(subject->ptr,entry->zi); 173 174 /* When we insert after the current element, but the current element 175 * is the tail of the list, we need to do a push. */ 176 if (next == NULL) { 177 subject->ptr = ziplistPush(subject->ptr,value->ptr,sdslen(value->ptr),REDIS_TAIL); 178 } else { 179 subject->ptr = ziplistInsert(subject->ptr,next,value->ptr,sdslen(value->ptr)); 180 } 181 } else { 182 subject->ptr = ziplistInsert(subject->ptr,entry->zi,value->ptr,sdslen(value->ptr)); 183 } 184 decrRefCount(value); 185 } else if (entry->li->encoding == REDIS_ENCODING_LINKEDLIST) { 186 if (where == REDIS_TAIL) { 187 listInsertNode(subject->ptr,entry->ln,value,AL_START_TAIL); 188 } else { 189 listInsertNode(subject->ptr,entry->ln,value,AL_START_HEAD); 190 } 191 incrRefCount(value); 192 } else { 193 redisPanic("Unknown list encoding"); 194 } 195 } 196 197 /* Compare the given object with the entry at the current position. */ 198 int listTypeEqual(listTypeEntry *entry, robj *o) { 199 listTypeIterator *li = entry->li; 200 if (li->encoding == REDIS_ENCODING_ZIPLIST) { 201 redisAssertWithInfo(NULL,o,o->encoding == REDIS_ENCODING_RAW); 202 return ziplistCompare(entry->zi,o->ptr,sdslen(o->ptr)); 203 } else if (li->encoding == REDIS_ENCODING_LINKEDLIST) { 204 return equalStringObjects(o,listNodeValue(entry->ln)); 205 } else { 206 redisPanic("Unknown list encoding"); 207 } 208 } 209 210 /* Delete the element pointed to. */ 211 void listTypeDelete(listTypeEntry *entry) { 212 listTypeIterator *li = entry->li; 213 if (li->encoding == REDIS_ENCODING_ZIPLIST) { 214 unsigned char *p = entry->zi; 215 li->subject->ptr = ziplistDelete(li->subject->ptr,&p); 216 217 /* Update position of the iterator depending on the direction */ 218 if (li->direction == REDIS_TAIL) 219 li->zi = p; 220 else 221 li->zi = ziplistPrev(li->subject->ptr,p); 222 } else if (entry->li->encoding == REDIS_ENCODING_LINKEDLIST) { 223 listNode *next; 224 if (li->direction == REDIS_TAIL) 225 next = entry->ln->next; 226 else 227 next = entry->ln->prev; 228 listDelNode(li->subject->ptr,entry->ln); 229 li->ln = next; 230 } else { 231 redisPanic("Unknown list encoding"); 232 } 233 } 234 235 void listTypeConvert(robj *subject, int enc) { 236 listTypeIterator *li; 237 listTypeEntry entry; 238 redisAssertWithInfo(NULL,subject,subject->type == REDIS_LIST); 239 240 if (enc == REDIS_ENCODING_LINKEDLIST) { 241 list *l = listCreate(); 242 listSetFreeMethod(l,decrRefCount); 243 244 /* listTypeGet returns a robj with incremented refcount */ 245 li = listTypeInitIterator(subject,0,REDIS_TAIL); 246 while (listTypeNext(li,&entry)) listAddNodeTail(l,listTypeGet(&entry)); 247 listTypeReleaseIterator(li); 248 249 subject->encoding = REDIS_ENCODING_LINKEDLIST; 250 zfree(subject->ptr); 251 subject->ptr = l; 252 } else { 253 redisPanic("Unsupported list conversion"); 254 } 255 } 256 257 /*----------------------------------------------------------------------------- 258 * List Commands 259 *----------------------------------------------------------------------------*/ 260 261 void pushGenericCommand(redisClient *c, int where) { 262 int j, addlen = 0, pushed = 0; 263 robj *lobj = lookupKeyWrite(c->db,c->argv[1]); 264 int may_have_waiting_clients = (lobj == NULL); 265 266 if (lobj && lobj->type != REDIS_LIST) { 267 addReply(c,shared.wrongtypeerr); 268 return; 269 } 270 271 for (j = 2; j < c->argc; j++) { 272 c->argv[j] = tryObjectEncoding(c->argv[j]); 273 if (may_have_waiting_clients) { 274 if (handleClientsWaitingListPush(c,c->argv[1],c->argv[j])) { 275 addlen++; 276 continue; 277 } else { 278 may_have_waiting_clients = 0; 279 } 280 } 281 if (!lobj) { 282 lobj = createZiplistObject(); 283 dbAdd(c->db,c->argv[1],lobj); 284 } 285 listTypePush(lobj,c->argv[j],where); 286 pushed++; 287 } 288 addReplyLongLong(c,addlen + (lobj ? listTypeLength(lobj) : 0)); 289 if (pushed) signalModifiedKey(c->db,c->argv[1]); 290 server.dirty += pushed; 291 } 292 293 void lpushCommand(redisClient *c) { 294 pushGenericCommand(c,REDIS_HEAD); 295 } 296 297 void rpushCommand(redisClient *c) { 298 pushGenericCommand(c,REDIS_TAIL); 299 } 300 301 void pushxGenericCommand(redisClient *c, robj *refval, robj *val, int where) { 302 robj *subject; 303 listTypeIterator *iter; 304 listTypeEntry entry; 305 int inserted = 0; 306 307 if ((subject = lookupKeyReadOrReply(c,c->argv[1],shared.czero)) == NULL || 308 checkType(c,subject,REDIS_LIST)) return; 309 310 if (refval != NULL) { 311 /* Note: we expect refval to be string-encoded because it is *not* the 312 * last argument of the multi-bulk LINSERT. */ 313 redisAssertWithInfo(c,refval,refval->encoding == REDIS_ENCODING_RAW); 314 315 /* We're not sure if this value can be inserted yet, but we cannot 316 * convert the list inside the iterator. We don't want to loop over 317 * the list twice (once to see if the value can be inserted and once 318 * to do the actual insert), so we assume this value can be inserted 319 * and convert the ziplist to a regular list if necessary. */ 320 listTypeTryConversion(subject,val); 321 322 /* Seek refval from head to tail */ 323 iter = listTypeInitIterator(subject,0,REDIS_TAIL); 324 while (listTypeNext(iter,&entry)) { 325 if (listTypeEqual(&entry,refval)) { 326 listTypeInsert(&entry,val,where); 327 inserted = 1; 328 break; 329 } 330 } 331 listTypeReleaseIterator(iter); 332 333 if (inserted) { 334 /* Check if the length exceeds the ziplist length threshold. */ 335 if (subject->encoding == REDIS_ENCODING_ZIPLIST && 336 ziplistLen(subject->ptr) > server.list_max_ziplist_entries) 337 listTypeConvert(subject,REDIS_ENCODING_LINKEDLIST); 338 signalModifiedKey(c->db,c->argv[1]); 339 server.dirty++; 340 } else { 341 /* Notify client of a failed insert */ 342 addReply(c,shared.cnegone); 343 return; 344 } 345 } else { 346 listTypePush(subject,val,where); 347 signalModifiedKey(c->db,c->argv[1]); 348 server.dirty++; 349 } 350 351 addReplyLongLong(c,listTypeLength(subject)); 352 } 353 354 void lpushxCommand(redisClient *c) { 355 c->argv[2] = tryObjectEncoding(c->argv[2]); 356 pushxGenericCommand(c,NULL,c->argv[2],REDIS_HEAD); 357 } 358 359 void rpushxCommand(redisClient *c) { 360 c->argv[2] = tryObjectEncoding(c->argv[2]); 361 pushxGenericCommand(c,NULL,c->argv[2],REDIS_TAIL); 362 } 363 364 void linsertCommand(redisClient *c) { 365 c->argv[4] = tryObjectEncoding(c->argv[4]); 366 if (strcasecmp(c->argv[2]->ptr,"after") == 0) { 367 pushxGenericCommand(c,c->argv[3],c->argv[4],REDIS_TAIL); 368 } else if (strcasecmp(c->argv[2]->ptr,"before") == 0) { 369 pushxGenericCommand(c,c->argv[3],c->argv[4],REDIS_HEAD); 370 } else { 371 addReply(c,shared.syntaxerr); 372 } 373 } 374 375 void llenCommand(redisClient *c) { 376 robj *o = lookupKeyReadOrReply(c,c->argv[1],shared.czero); 377 if (o == NULL || checkType(c,o,REDIS_LIST)) return; 378 addReplyLongLong(c,listTypeLength(o)); 379 } 380 381 void lindexCommand(redisClient *c) { 382 robj *o = lookupKeyReadOrReply(c,c->argv[1],shared.nullbulk); 383 if (o == NULL || checkType(c,o,REDIS_LIST)) return; 384 long index; 385 robj *value = NULL; 386 387 if ((getLongFromObjectOrReply(c, c->argv[2], &index, NULL) != REDIS_OK)) 388 return; 389 390 if (o->encoding == REDIS_ENCODING_ZIPLIST) { 391 unsigned char *p; 392 unsigned char *vstr; 393 unsigned int vlen; 394 long long vlong; 395 p = ziplistIndex(o->ptr,index); 396 if (ziplistGet(p,&vstr,&vlen,&vlong)) { 397 if (vstr) { 398 value = createStringObject((char*)vstr,vlen); 399 } else { 400 value = createStringObjectFromLongLong(vlong); 401 } 402 addReplyBulk(c,value); 403 decrRefCount(value); 404 } else { 405 addReply(c,shared.nullbulk); 406 } 407 } else if (o->encoding == REDIS_ENCODING_LINKEDLIST) { 408 listNode *ln = listIndex(o->ptr,index); 409 if (ln != NULL) { 410 value = listNodeValue(ln); 411 addReplyBulk(c,value); 412 } else { 413 addReply(c,shared.nullbulk); 414 } 415 } else { 416 redisPanic("Unknown list encoding"); 417 } 418 } 419 420 void lsetCommand(redisClient *c) { 421 robj *o = lookupKeyWriteOrReply(c,c->argv[1],shared.nokeyerr); 422 if (o == NULL || checkType(c,o,REDIS_LIST)) return; 423 long index; 424 robj *value = (c->argv[3] = tryObjectEncoding(c->argv[3])); 425 426 if ((getLongFromObjectOrReply(c, c->argv[2], &index, NULL) != REDIS_OK)) 427 return; 428 429 listTypeTryConversion(o,value); 430 if (o->encoding == REDIS_ENCODING_ZIPLIST) { 431 unsigned char *p, *zl = o->ptr; 432 p = ziplistIndex(zl,index); 433 if (p == NULL) { 434 addReply(c,shared.outofrangeerr); 435 } else { 436 o->ptr = ziplistDelete(o->ptr,&p); 437 value = getDecodedObject(value); 438 o->ptr = ziplistInsert(o->ptr,p,value->ptr,sdslen(value->ptr)); 439 decrRefCount(value); 440 addReply(c,shared.ok); 441 signalModifiedKey(c->db,c->argv[1]); 442 server.dirty++; 443 } 444 } else if (o->encoding == REDIS_ENCODING_LINKEDLIST) { 445 listNode *ln = listIndex(o->ptr,index); 446 if (ln == NULL) { 447 addReply(c,shared.outofrangeerr); 448 } else { 449 decrRefCount((robj*)listNodeValue(ln)); 450 listNodeValue(ln) = value; 451 incrRefCount(value); 452 addReply(c,shared.ok); 453 signalModifiedKey(c->db,c->argv[1]); 454 server.dirty++; 455 } 456 } else { 457 redisPanic("Unknown list encoding"); 458 } 459 } 460 461 void popGenericCommand(redisClient *c, int where) { 462 robj *o = lookupKeyWriteOrReply(c,c->argv[1],shared.nullbulk); 463 if (o == NULL || checkType(c,o,REDIS_LIST)) return; 464 465 robj *value = listTypePop(o,where); 466 if (value == NULL) { 467 addReply(c,shared.nullbulk); 468 } else { 469 addReplyBulk(c,value); 470 decrRefCount(value); 471 if (listTypeLength(o) == 0) dbDelete(c->db,c->argv[1]); 472 signalModifiedKey(c->db,c->argv[1]); 473 server.dirty++; 474 } 475 } 476 477 void lpopCommand(redisClient *c) { 478 popGenericCommand(c,REDIS_HEAD); 479 } 480 481 void rpopCommand(redisClient *c) { 482 popGenericCommand(c,REDIS_TAIL); 483 } 484 485 void lrangeCommand(redisClient *c) { 486 robj *o; 487 long start; 488 long end; 489 int llen; 490 int rangelen; 491 492 if ((getLongFromObjectOrReply(c, c->argv[2], &start, NULL) != REDIS_OK) || 493 (getLongFromObjectOrReply(c, c->argv[3], &end, NULL) != REDIS_OK)) return; 494 495 if ((o = lookupKeyReadOrReply(c,c->argv[1],shared.emptymultibulk)) == NULL 496 || checkType(c,o,REDIS_LIST)) return; 497 llen = listTypeLength(o); 498 499 /* convert negative indexes */ 500 if (start < 0) start = llen+start; 501 if (end < 0) end = llen+end; 502 if (start < 0) start = 0; 503 504 /* Invariant: start >= 0, so this test will be true when end < 0. 505 * The range is empty when start > end or start >= length. */ 506 if (start > end || start >= llen) { 507 addReply(c,shared.emptymultibulk); 508 return; 509 } 510 if (end >= llen) end = llen-1; 511 rangelen = (end-start)+1; 512 513 /* Return the result in form of a multi-bulk reply */ 514 addReplyMultiBulkLen(c,rangelen); 515 if (o->encoding == REDIS_ENCODING_ZIPLIST) { 516 unsigned char *p = ziplistIndex(o->ptr,start); 517 unsigned char *vstr; 518 unsigned int vlen; 519 long long vlong; 520 521 while(rangelen--) { 522 ziplistGet(p,&vstr,&vlen,&vlong); 523 if (vstr) { 524 addReplyBulkCBuffer(c,vstr,vlen); 525 } else { 526 addReplyBulkLongLong(c,vlong); 527 } 528 p = ziplistNext(o->ptr,p); 529 } 530 } else if (o->encoding == REDIS_ENCODING_LINKEDLIST) { 531 listNode *ln; 532 533 /* If we are nearest to the end of the list, reach the element 534 * starting from tail and going backward, as it is faster. */ 535 if (start > llen/2) start -= llen; 536 ln = listIndex(o->ptr,start); 537 538 while(rangelen--) { 539 addReplyBulk(c,ln->value); 540 ln = ln->next; 541 } 542 } else { 543 redisPanic("List encoding is not LINKEDLIST nor ZIPLIST!"); 544 } 545 } 546 547 void ltrimCommand(redisClient *c) { 548 robj *o; 549 long start; 550 long end; 551 int llen; 552 int j, ltrim, rtrim; 553 list *list; 554 listNode *ln; 555 556 if ((getLongFromObjectOrReply(c, c->argv[2], &start, NULL) != REDIS_OK) || 557 (getLongFromObjectOrReply(c, c->argv[3], &end, NULL) != REDIS_OK)) return; 558 559 if ((o = lookupKeyWriteOrReply(c,c->argv[1],shared.ok)) == NULL || 560 checkType(c,o,REDIS_LIST)) return; 561 llen = listTypeLength(o); 562 563 /* convert negative indexes */ 564 if (start < 0) start = llen+start; 565 if (end < 0) end = llen+end; 566 if (start < 0) start = 0; 567 568 /* Invariant: start >= 0, so this test will be true when end < 0. 569 * The range is empty when start > end or start >= length. */ 570 if (start > end || start >= llen) { 571 /* Out of range start or start > end result in empty list */ 572 ltrim = llen; 573 rtrim = 0; 574 } else { 575 if (end >= llen) end = llen-1; 576 ltrim = start; 577 rtrim = llen-end-1; 578 } 579 580 /* Remove list elements to perform the trim */ 581 if (o->encoding == REDIS_ENCODING_ZIPLIST) { 582 o->ptr = ziplistDeleteRange(o->ptr,0,ltrim); 583 o->ptr = ziplistDeleteRange(o->ptr,-rtrim,rtrim); 584 } else if (o->encoding == REDIS_ENCODING_LINKEDLIST) { 585 list = o->ptr; 586 for (j = 0; j < ltrim; j++) { 587 ln = listFirst(list); 588 listDelNode(list,ln); 589 } 590 for (j = 0; j < rtrim; j++) { 591 ln = listLast(list); 592 listDelNode(list,ln); 593 } 594 } else { 595 redisPanic("Unknown list encoding"); 596 } 597 if (listTypeLength(o) == 0) dbDelete(c->db,c->argv[1]); 598 signalModifiedKey(c->db,c->argv[1]); 599 server.dirty++; 600 addReply(c,shared.ok); 601 } 602 603 void lremCommand(redisClient *c) { 604 robj *subject, *obj; 605 obj = c->argv[3] = tryObjectEncoding(c->argv[3]); 606 long toremove; 607 int removed = 0; 608 listTypeEntry entry; 609 610 if ((getLongFromObjectOrReply(c, c->argv[2], &toremove, NULL) != REDIS_OK)) 611 return; 612 613 subject = lookupKeyWriteOrReply(c,c->argv[1],shared.czero); 614 if (subject == NULL || checkType(c,subject,REDIS_LIST)) return; 615 616 /* Make sure obj is raw when we're dealing with a ziplist */ 617 if (subject->encoding == REDIS_ENCODING_ZIPLIST) 618 obj = getDecodedObject(obj); 619 620 listTypeIterator *li; 621 if (toremove < 0) { 622 toremove = -toremove; 623 li = listTypeInitIterator(subject,-1,REDIS_HEAD); 624 } else { 625 li = listTypeInitIterator(subject,0,REDIS_TAIL); 626 } 627 628 while (listTypeNext(li,&entry)) { 629 if (listTypeEqual(&entry,obj)) { 630 listTypeDelete(&entry); 631 server.dirty++; 632 removed++; 633 if (toremove && removed == toremove) break; 634 } 635 } 636 listTypeReleaseIterator(li); 637 638 /* Clean up raw encoded object */ 639 if (subject->encoding == REDIS_ENCODING_ZIPLIST) 640 decrRefCount(obj); 641 642 if (listTypeLength(subject) == 0) dbDelete(c->db,c->argv[1]); 643 addReplyLongLong(c,removed); 644 if (removed) signalModifiedKey(c->db,c->argv[1]); 645 } 646 647 /* This is the semantic of this command: 648 * RPOPLPUSH srclist dstlist: 649 * IF LLEN(srclist) > 0 650 * element = RPOP srclist 651 * LPUSH dstlist element 652 * RETURN element 653 * ELSE 654 * RETURN nil 655 * END 656 * END 657 * 658 * The idea is to be able to get an element from a list in a reliable way 659 * since the element is not just returned but pushed against another list 660 * as well. This command was originally proposed by Ezra Zygmuntowicz. 661 */ 662 663 void rpoplpushHandlePush(redisClient *origclient, redisClient *c, robj *dstkey, robj *dstobj, robj *value) { 664 robj *aux; 665 666 if (!handleClientsWaitingListPush(origclient,dstkey,value)) { 667 /* Create the list if the key does not exist */ 668 if (!dstobj) { 669 dstobj = createZiplistObject(); 670 dbAdd(c->db,dstkey,dstobj); 671 } else { 672 signalModifiedKey(c->db,dstkey); 673 } 674 listTypePush(dstobj,value,REDIS_HEAD); 675 /* If we are pushing as a result of LPUSH against a key 676 * watched by BRPOPLPUSH, we need to rewrite the command vector 677 * as an LPUSH. 678 * 679 * If this is called directly by RPOPLPUSH (either directly 680 * or via a BRPOPLPUSH where the popped list exists) 681 * we should replicate the RPOPLPUSH command itself. */ 682 if (c != origclient) { 683 aux = createStringObject("LPUSH",5); 684 rewriteClientCommandVector(origclient,3,aux,dstkey,value); 685 decrRefCount(aux); 686 } else { 687 /* Make sure to always use RPOPLPUSH in the replication / AOF, 688 * even if the original command was BRPOPLPUSH. */ 689 aux = createStringObject("RPOPLPUSH",9); 690 rewriteClientCommandVector(origclient,3,aux,c->argv[1],c->argv[2]); 691 decrRefCount(aux); 692 } 693 server.dirty++; 694 } 695 696 /* Always send the pushed value to the client. */ 697 addReplyBulk(c,value); 698 } 699 700 void rpoplpushCommand(redisClient *c) { 701 robj *sobj, *value; 702 if ((sobj = lookupKeyWriteOrReply(c,c->argv[1],shared.nullbulk)) == NULL || 703 checkType(c,sobj,REDIS_LIST)) return; 704 705 if (listTypeLength(sobj) == 0) { 706 addReply(c,shared.nullbulk); 707 } else { 708 robj *dobj = lookupKeyWrite(c->db,c->argv[2]); 709 robj *touchedkey = c->argv[1]; 710 711 if (dobj && checkType(c,dobj,REDIS_LIST)) return; 712 value = listTypePop(sobj,REDIS_TAIL); 713 /* We saved touched key, and protect it, since rpoplpushHandlePush 714 * may change the client command argument vector. */ 715 incrRefCount(touchedkey); 716 rpoplpushHandlePush(c,c,c->argv[2],dobj,value); 717 718 /* listTypePop returns an object with its refcount incremented */ 719 decrRefCount(value); 720 721 /* Delete the source list when it is empty */ 722 if (listTypeLength(sobj) == 0) dbDelete(c->db,touchedkey); 723 signalModifiedKey(c->db,touchedkey); 724 decrRefCount(touchedkey); 725 server.dirty++; 726 } 727 } 728 729 /*----------------------------------------------------------------------------- 730 * Blocking POP operations 731 *----------------------------------------------------------------------------*/ 732 733 /* Currently Redis blocking operations support is limited to list POP ops, 734 * so the current implementation is not fully generic, but it is also not 735 * completely specific so it will not require a rewrite to support new 736 * kind of blocking operations in the future. 737 * 738 * Still it's important to note that list blocking operations can be already 739 * used as a notification mechanism in order to implement other blocking 740 * operations at application level, so there must be a very strong evidence 741 * of usefulness and generality before new blocking operations are implemented. 742 * 743 * This is how the current blocking POP works, we use BLPOP as example: 744 * - If the user calls BLPOP and the key exists and contains a non empty list 745 * then LPOP is called instead. So BLPOP is semantically the same as LPOP 746 * if there is not to block. 747 * - If instead BLPOP is called and the key does not exists or the list is 748 * empty we need to block. In order to do so we remove the notification for 749 * new data to read in the client socket (so that we'll not serve new 750 * requests if the blocking request is not served). Also we put the client 751 * in a dictionary (db->blocking_keys) mapping keys to a list of clients 752 * blocking for this keys. 753 * - If a PUSH operation against a key with blocked clients waiting is 754 * performed, we serve the first in the list: basically instead to push 755 * the new element inside the list we return it to the (first / oldest) 756 * blocking client, unblock the client, and remove it form the list. 757 * 758 * The above comment and the source code should be enough in order to understand 759 * the implementation and modify / fix it later. 760 */ 761 762 /* Set a client in blocking mode for the specified key, with the specified 763 * timeout */ 764 void blockForKeys(redisClient *c, robj **keys, int numkeys, time_t timeout, robj *target) { 765 dictEntry *de; 766 list *l; 767 int j; 768 769 c->bpop.keys = zmalloc(sizeof(robj*)*numkeys); 770 c->bpop.count = numkeys; 771 c->bpop.timeout = timeout; 772 c->bpop.target = target; 773 774 if (target != NULL) { 775 incrRefCount(target); 776 } 777 778 for (j = 0; j < numkeys; j++) { 779 /* Add the key in the client structure, to map clients -> keys */ 780 c->bpop.keys[j] = keys[j]; 781 incrRefCount(keys[j]); 782 783 /* And in the other "side", to map keys -> clients */ 784 de = dictFind(c->db->blocking_keys,keys[j]); 785 if (de == NULL) { 786 int retval; 787 788 /* For every key we take a list of clients blocked for it */ 789 l = listCreate(); 790 retval = dictAdd(c->db->blocking_keys,keys[j],l); 791 incrRefCount(keys[j]); 792 redisAssertWithInfo(c,keys[j],retval == DICT_OK); 793 } else { 794 l = dictGetVal(de); 795 } 796 listAddNodeTail(l,c); 797 } 798 /* Mark the client as a blocked client */ 799 c->flags |= REDIS_BLOCKED; 800 server.bpop_blocked_clients++; 801 } 802 803 /* Unblock a client that's waiting in a blocking operation such as BLPOP */ 804 void unblockClientWaitingData(redisClient *c) { 805 dictEntry *de; 806 list *l; 807 int j; 808 809 redisAssertWithInfo(c,NULL,c->bpop.keys != NULL); 810 /* The client may wait for multiple keys, so unblock it for every key. */ 811 for (j = 0; j < c->bpop.count; j++) { 812 /* Remove this client from the list of clients waiting for this key. */ 813 de = dictFind(c->db->blocking_keys,c->bpop.keys[j]); 814 redisAssertWithInfo(c,c->bpop.keys[j],de != NULL); 815 l = dictGetVal(de); 816 listDelNode(l,listSearchKey(l,c)); 817 /* If the list is empty we need to remove it to avoid wasting memory */ 818 if (listLength(l) == 0) 819 dictDelete(c->db->blocking_keys,c->bpop.keys[j]); 820 decrRefCount(c->bpop.keys[j]); 821 } 822 823 /* Cleanup the client structure */ 824 zfree(c->bpop.keys); 825 c->bpop.keys = NULL; 826 if (c->bpop.target) decrRefCount(c->bpop.target); 827 c->bpop.target = NULL; 828 c->flags &= ~REDIS_BLOCKED; 829 c->flags |= REDIS_UNBLOCKED; 830 server.bpop_blocked_clients--; 831 listAddNodeTail(server.unblocked_clients,c); 832 } 833 834 /* This should be called from any function PUSHing into lists. 835 * 'c' is the "pushing client", 'key' is the key it is pushing data against, 836 * 'ele' is the element pushed. 837 * 838 * If the function returns 0 there was no client waiting for a list push 839 * against this key. 840 * 841 * If the function returns 1 there was a client waiting for a list push 842 * against this key, the element was passed to this client thus it's not 843 * needed to actually add it to the list and the caller should return asap. */ 844 int handleClientsWaitingListPush(redisClient *c, robj *key, robj *ele) { 845 struct dictEntry *de; 846 redisClient *receiver; 847 int numclients; 848 list *clients; 849 listNode *ln; 850 robj *dstkey, *dstobj; 851 852 de = dictFind(c->db->blocking_keys,key); 853 if (de == NULL) return 0; 854 clients = dictGetVal(de); 855 numclients = listLength(clients); 856 857 /* Try to handle the push as long as there are clients waiting for a push. 858 * Note that "numclients" is used because the list of clients waiting for a 859 * push on "key" is deleted by unblockClient() when empty. 860 * 861 * This loop will have more than 1 iteration when there is a BRPOPLPUSH 862 * that cannot push the target list because it does not contain a list. If 863 * this happens, it simply tries the next client waiting for a push. */ 864 while (numclients--) { 865 ln = listFirst(clients); 866 redisAssertWithInfo(c,key,ln != NULL); 867 receiver = ln->value; 868 dstkey = receiver->bpop.target; 869 870 /* Protect receiver->bpop.target, that will be freed by 871 * the next unblockClientWaitingData() call. */ 872 if (dstkey) incrRefCount(dstkey); 873 874 /* This should remove the first element of the "clients" list. */ 875 unblockClientWaitingData(receiver); 876 877 if (dstkey == NULL) { 878 /* BRPOP/BLPOP */ 879 addReplyMultiBulkLen(receiver,2); 880 addReplyBulk(receiver,key); 881 addReplyBulk(receiver,ele); 882 return 1; /* Serve just the first client as in B[RL]POP semantics */ 883 } else { 884 /* BRPOPLPUSH, note that receiver->db is always equal to c->db. */ 885 dstobj = lookupKeyWrite(receiver->db,dstkey); 886 if (!(dstobj && checkType(receiver,dstobj,REDIS_LIST))) { 887 rpoplpushHandlePush(c,receiver,dstkey,dstobj,ele); 888 decrRefCount(dstkey); 889 return 1; 890 } 891 decrRefCount(dstkey); 892 } 893 } 894 895 return 0; 896 } 897 898 int getTimeoutFromObjectOrReply(redisClient *c, robj *object, time_t *timeout) { 899 long tval; 900 901 if (getLongFromObjectOrReply(c,object,&tval, 902 "timeout is not an integer or out of range") != REDIS_OK) 903 return REDIS_ERR; 904 905 if (tval < 0) { 906 addReplyError(c,"timeout is negative"); 907 return REDIS_ERR; 908 } 909 910 if (tval > 0) tval += time(NULL); 911 *timeout = tval; 912 913 return REDIS_OK; 914 } 915 916 /* Blocking RPOP/LPOP */ 917 void blockingPopGenericCommand(redisClient *c, int where) { 918 robj *o; 919 time_t timeout; 920 int j; 921 922 if (getTimeoutFromObjectOrReply(c,c->argv[c->argc-1],&timeout) != REDIS_OK) 923 return; 924 925 for (j = 1; j < c->argc-1; j++) { 926 o = lookupKeyWrite(c->db,c->argv[j]); 927 if (o != NULL) { 928 if (o->type != REDIS_LIST) { 929 addReply(c,shared.wrongtypeerr); 930 return; 931 } else { 932 if (listTypeLength(o) != 0) { 933 /* If the list contains elements fall back to the usual 934 * non-blocking POP operation */ 935 struct redisCommand *orig_cmd; 936 robj *argv[2], **orig_argv; 937 int orig_argc; 938 939 /* We need to alter the command arguments before to call 940 * popGenericCommand() as the command takes a single key. */ 941 orig_argv = c->argv; 942 orig_argc = c->argc; 943 orig_cmd = c->cmd; 944 argv[1] = c->argv[j]; 945 c->argv = argv; 946 c->argc = 2; 947 948 /* Also the return value is different, we need to output 949 * the multi bulk reply header and the key name. The 950 * "real" command will add the last element (the value) 951 * for us. If this souds like an hack to you it's just 952 * because it is... */ 953 addReplyMultiBulkLen(c,2); 954 addReplyBulk(c,argv[1]); 955 956 popGenericCommand(c,where); 957 958 /* Fix the client structure with the original stuff */ 959 c->argv = orig_argv; 960 c->argc = orig_argc; 961 c->cmd = orig_cmd; 962 963 return; 964 } 965 } 966 } 967 } 968 969 /* If we are inside a MULTI/EXEC and the list is empty the only thing 970 * we can do is treating it as a timeout (even with timeout 0). */ 971 if (c->flags & REDIS_MULTI) { 972 addReply(c,shared.nullmultibulk); 973 return; 974 } 975 976 /* If the list is empty or the key does not exists we must block */ 977 blockForKeys(c, c->argv + 1, c->argc - 2, timeout, NULL); 978 } 979 980 void blpopCommand(redisClient *c) { 981 blockingPopGenericCommand(c,REDIS_HEAD); 982 } 983 984 void brpopCommand(redisClient *c) { 985 blockingPopGenericCommand(c,REDIS_TAIL); 986 } 987 988 void brpoplpushCommand(redisClient *c) { 989 time_t timeout; 990 991 if (getTimeoutFromObjectOrReply(c,c->argv[3],&timeout) != REDIS_OK) 992 return; 993 994 robj *key = lookupKeyWrite(c->db, c->argv[1]); 995 996 if (key == NULL) { 997 if (c->flags & REDIS_MULTI) { 998 999 /* Blocking against an empty list in a multi state 1000 * returns immediately. */ 1001 addReply(c, shared.nullbulk); 1002 } else { 1003 /* The list is empty and the client blocks. */ 1004 blockForKeys(c, c->argv + 1, 1, timeout, c->argv[2]); 1005 } 1006 } else { 1007 if (key->type != REDIS_LIST) { 1008 addReply(c, shared.wrongtypeerr); 1009 } else { 1010 1011 /* The list exists and has elements, so 1012 * the regular rpoplpushCommand is executed. */ 1013 redisAssertWithInfo(c,key,listTypeLength(key) > 0); 1014 rpoplpushCommand(c); 1015 } 1016 } 1017 } 1018