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, long 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, waiting = 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 waiting++; 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, waiting + (lobj ? listTypeLength(lobj) : 0)); 289 if (pushed) signalModifiedKey(c->db,c->argv[1]); 290 server.dirty += pushed; 291 292 /* Alter the replication of the command accordingly to the number of 293 * list elements delivered to clients waiting into a blocking operation. 294 * We do that only if there were waiting clients, and only if still some 295 * element was pushed into the list (othewise dirty is 0 and nothign will 296 * be propagated). */ 297 if (waiting && pushed) { 298 /* CMD KEY a b C D E */ 299 for (j = 0; j < waiting; j++) decrRefCount(c->argv[j+2]); 300 memmove(c->argv+2,c->argv+2+waiting,sizeof(robj*)*pushed); 301 c->argc -= waiting; 302 } 303 } 304 305 void lpushCommand(redisClient *c) { 306 pushGenericCommand(c,REDIS_HEAD); 307 } 308 309 void rpushCommand(redisClient *c) { 310 pushGenericCommand(c,REDIS_TAIL); 311 } 312 313 void pushxGenericCommand(redisClient *c, robj *refval, robj *val, int where) { 314 robj *subject; 315 listTypeIterator *iter; 316 listTypeEntry entry; 317 int inserted = 0; 318 319 if ((subject = lookupKeyReadOrReply(c,c->argv[1],shared.czero)) == NULL || 320 checkType(c,subject,REDIS_LIST)) return; 321 322 if (refval != NULL) { 323 /* Note: we expect refval to be string-encoded because it is *not* the 324 * last argument of the multi-bulk LINSERT. */ 325 redisAssertWithInfo(c,refval,refval->encoding == REDIS_ENCODING_RAW); 326 327 /* We're not sure if this value can be inserted yet, but we cannot 328 * convert the list inside the iterator. We don't want to loop over 329 * the list twice (once to see if the value can be inserted and once 330 * to do the actual insert), so we assume this value can be inserted 331 * and convert the ziplist to a regular list if necessary. */ 332 listTypeTryConversion(subject,val); 333 334 /* Seek refval from head to tail */ 335 iter = listTypeInitIterator(subject,0,REDIS_TAIL); 336 while (listTypeNext(iter,&entry)) { 337 if (listTypeEqual(&entry,refval)) { 338 listTypeInsert(&entry,val,where); 339 inserted = 1; 340 break; 341 } 342 } 343 listTypeReleaseIterator(iter); 344 345 if (inserted) { 346 /* Check if the length exceeds the ziplist length threshold. */ 347 if (subject->encoding == REDIS_ENCODING_ZIPLIST && 348 ziplistLen(subject->ptr) > server.list_max_ziplist_entries) 349 listTypeConvert(subject,REDIS_ENCODING_LINKEDLIST); 350 signalModifiedKey(c->db,c->argv[1]); 351 server.dirty++; 352 } else { 353 /* Notify client of a failed insert */ 354 addReply(c,shared.cnegone); 355 return; 356 } 357 } else { 358 listTypePush(subject,val,where); 359 signalModifiedKey(c->db,c->argv[1]); 360 server.dirty++; 361 } 362 363 addReplyLongLong(c,listTypeLength(subject)); 364 } 365 366 void lpushxCommand(redisClient *c) { 367 c->argv[2] = tryObjectEncoding(c->argv[2]); 368 pushxGenericCommand(c,NULL,c->argv[2],REDIS_HEAD); 369 } 370 371 void rpushxCommand(redisClient *c) { 372 c->argv[2] = tryObjectEncoding(c->argv[2]); 373 pushxGenericCommand(c,NULL,c->argv[2],REDIS_TAIL); 374 } 375 376 void linsertCommand(redisClient *c) { 377 c->argv[4] = tryObjectEncoding(c->argv[4]); 378 if (strcasecmp(c->argv[2]->ptr,"after") == 0) { 379 pushxGenericCommand(c,c->argv[3],c->argv[4],REDIS_TAIL); 380 } else if (strcasecmp(c->argv[2]->ptr,"before") == 0) { 381 pushxGenericCommand(c,c->argv[3],c->argv[4],REDIS_HEAD); 382 } else { 383 addReply(c,shared.syntaxerr); 384 } 385 } 386 387 void llenCommand(redisClient *c) { 388 robj *o = lookupKeyReadOrReply(c,c->argv[1],shared.czero); 389 if (o == NULL || checkType(c,o,REDIS_LIST)) return; 390 addReplyLongLong(c,listTypeLength(o)); 391 } 392 393 void lindexCommand(redisClient *c) { 394 robj *o = lookupKeyReadOrReply(c,c->argv[1],shared.nullbulk); 395 if (o == NULL || checkType(c,o,REDIS_LIST)) return; 396 long index; 397 robj *value = NULL; 398 399 if ((getLongFromObjectOrReply(c, c->argv[2], &index, NULL) != REDIS_OK)) 400 return; 401 402 if (o->encoding == REDIS_ENCODING_ZIPLIST) { 403 unsigned char *p; 404 unsigned char *vstr; 405 unsigned int vlen; 406 long long vlong; 407 p = ziplistIndex(o->ptr,index); 408 if (ziplistGet(p,&vstr,&vlen,&vlong)) { 409 if (vstr) { 410 value = createStringObject((char*)vstr,vlen); 411 } else { 412 value = createStringObjectFromLongLong(vlong); 413 } 414 addReplyBulk(c,value); 415 decrRefCount(value); 416 } else { 417 addReply(c,shared.nullbulk); 418 } 419 } else if (o->encoding == REDIS_ENCODING_LINKEDLIST) { 420 listNode *ln = listIndex(o->ptr,index); 421 if (ln != NULL) { 422 value = listNodeValue(ln); 423 addReplyBulk(c,value); 424 } else { 425 addReply(c,shared.nullbulk); 426 } 427 } else { 428 redisPanic("Unknown list encoding"); 429 } 430 } 431 432 void lsetCommand(redisClient *c) { 433 robj *o = lookupKeyWriteOrReply(c,c->argv[1],shared.nokeyerr); 434 if (o == NULL || checkType(c,o,REDIS_LIST)) return; 435 long index; 436 robj *value = (c->argv[3] = tryObjectEncoding(c->argv[3])); 437 438 if ((getLongFromObjectOrReply(c, c->argv[2], &index, NULL) != REDIS_OK)) 439 return; 440 441 listTypeTryConversion(o,value); 442 if (o->encoding == REDIS_ENCODING_ZIPLIST) { 443 unsigned char *p, *zl = o->ptr; 444 p = ziplistIndex(zl,index); 445 if (p == NULL) { 446 addReply(c,shared.outofrangeerr); 447 } else { 448 o->ptr = ziplistDelete(o->ptr,&p); 449 value = getDecodedObject(value); 450 o->ptr = ziplistInsert(o->ptr,p,value->ptr,sdslen(value->ptr)); 451 decrRefCount(value); 452 addReply(c,shared.ok); 453 signalModifiedKey(c->db,c->argv[1]); 454 server.dirty++; 455 } 456 } else if (o->encoding == REDIS_ENCODING_LINKEDLIST) { 457 listNode *ln = listIndex(o->ptr,index); 458 if (ln == NULL) { 459 addReply(c,shared.outofrangeerr); 460 } else { 461 decrRefCount((robj*)listNodeValue(ln)); 462 listNodeValue(ln) = value; 463 incrRefCount(value); 464 addReply(c,shared.ok); 465 signalModifiedKey(c->db,c->argv[1]); 466 server.dirty++; 467 } 468 } else { 469 redisPanic("Unknown list encoding"); 470 } 471 } 472 473 void popGenericCommand(redisClient *c, int where) { 474 robj *o = lookupKeyWriteOrReply(c,c->argv[1],shared.nullbulk); 475 if (o == NULL || checkType(c,o,REDIS_LIST)) return; 476 477 robj *value = listTypePop(o,where); 478 if (value == NULL) { 479 addReply(c,shared.nullbulk); 480 } else { 481 addReplyBulk(c,value); 482 decrRefCount(value); 483 if (listTypeLength(o) == 0) dbDelete(c->db,c->argv[1]); 484 signalModifiedKey(c->db,c->argv[1]); 485 server.dirty++; 486 } 487 } 488 489 void lpopCommand(redisClient *c) { 490 popGenericCommand(c,REDIS_HEAD); 491 } 492 493 void rpopCommand(redisClient *c) { 494 popGenericCommand(c,REDIS_TAIL); 495 } 496 497 void lrangeCommand(redisClient *c) { 498 robj *o; 499 long start, end, llen, rangelen; 500 501 if ((getLongFromObjectOrReply(c, c->argv[2], &start, NULL) != REDIS_OK) || 502 (getLongFromObjectOrReply(c, c->argv[3], &end, NULL) != REDIS_OK)) return; 503 504 if ((o = lookupKeyReadOrReply(c,c->argv[1],shared.emptymultibulk)) == NULL 505 || checkType(c,o,REDIS_LIST)) return; 506 llen = listTypeLength(o); 507 508 /* convert negative indexes */ 509 if (start < 0) start = llen+start; 510 if (end < 0) end = llen+end; 511 if (start < 0) start = 0; 512 513 /* Invariant: start >= 0, so this test will be true when end < 0. 514 * The range is empty when start > end or start >= length. */ 515 if (start > end || start >= llen) { 516 addReply(c,shared.emptymultibulk); 517 return; 518 } 519 if (end >= llen) end = llen-1; 520 rangelen = (end-start)+1; 521 522 /* Return the result in form of a multi-bulk reply */ 523 addReplyMultiBulkLen(c,rangelen); 524 if (o->encoding == REDIS_ENCODING_ZIPLIST) { 525 unsigned char *p = ziplistIndex(o->ptr,start); 526 unsigned char *vstr; 527 unsigned int vlen; 528 long long vlong; 529 530 while(rangelen--) { 531 ziplistGet(p,&vstr,&vlen,&vlong); 532 if (vstr) { 533 addReplyBulkCBuffer(c,vstr,vlen); 534 } else { 535 addReplyBulkLongLong(c,vlong); 536 } 537 p = ziplistNext(o->ptr,p); 538 } 539 } else if (o->encoding == REDIS_ENCODING_LINKEDLIST) { 540 listNode *ln; 541 542 /* If we are nearest to the end of the list, reach the element 543 * starting from tail and going backward, as it is faster. */ 544 if (start > llen/2) start -= llen; 545 ln = listIndex(o->ptr,start); 546 547 while(rangelen--) { 548 addReplyBulk(c,ln->value); 549 ln = ln->next; 550 } 551 } else { 552 redisPanic("List encoding is not LINKEDLIST nor ZIPLIST!"); 553 } 554 } 555 556 void ltrimCommand(redisClient *c) { 557 robj *o; 558 long start, end, llen, j, ltrim, rtrim; 559 list *list; 560 listNode *ln; 561 562 if ((getLongFromObjectOrReply(c, c->argv[2], &start, NULL) != REDIS_OK) || 563 (getLongFromObjectOrReply(c, c->argv[3], &end, NULL) != REDIS_OK)) return; 564 565 if ((o = lookupKeyWriteOrReply(c,c->argv[1],shared.ok)) == NULL || 566 checkType(c,o,REDIS_LIST)) return; 567 llen = listTypeLength(o); 568 569 /* convert negative indexes */ 570 if (start < 0) start = llen+start; 571 if (end < 0) end = llen+end; 572 if (start < 0) start = 0; 573 574 /* Invariant: start >= 0, so this test will be true when end < 0. 575 * The range is empty when start > end or start >= length. */ 576 if (start > end || start >= llen) { 577 /* Out of range start or start > end result in empty list */ 578 ltrim = llen; 579 rtrim = 0; 580 } else { 581 if (end >= llen) end = llen-1; 582 ltrim = start; 583 rtrim = llen-end-1; 584 } 585 586 /* Remove list elements to perform the trim */ 587 if (o->encoding == REDIS_ENCODING_ZIPLIST) { 588 o->ptr = ziplistDeleteRange(o->ptr,0,ltrim); 589 o->ptr = ziplistDeleteRange(o->ptr,-rtrim,rtrim); 590 } else if (o->encoding == REDIS_ENCODING_LINKEDLIST) { 591 list = o->ptr; 592 for (j = 0; j < ltrim; j++) { 593 ln = listFirst(list); 594 listDelNode(list,ln); 595 } 596 for (j = 0; j < rtrim; j++) { 597 ln = listLast(list); 598 listDelNode(list,ln); 599 } 600 } else { 601 redisPanic("Unknown list encoding"); 602 } 603 if (listTypeLength(o) == 0) dbDelete(c->db,c->argv[1]); 604 signalModifiedKey(c->db,c->argv[1]); 605 server.dirty++; 606 addReply(c,shared.ok); 607 } 608 609 void lremCommand(redisClient *c) { 610 robj *subject, *obj; 611 obj = c->argv[3] = tryObjectEncoding(c->argv[3]); 612 long toremove; 613 long removed = 0; 614 listTypeEntry entry; 615 616 if ((getLongFromObjectOrReply(c, c->argv[2], &toremove, NULL) != REDIS_OK)) 617 return; 618 619 subject = lookupKeyWriteOrReply(c,c->argv[1],shared.czero); 620 if (subject == NULL || checkType(c,subject,REDIS_LIST)) return; 621 622 /* Make sure obj is raw when we're dealing with a ziplist */ 623 if (subject->encoding == REDIS_ENCODING_ZIPLIST) 624 obj = getDecodedObject(obj); 625 626 listTypeIterator *li; 627 if (toremove < 0) { 628 toremove = -toremove; 629 li = listTypeInitIterator(subject,-1,REDIS_HEAD); 630 } else { 631 li = listTypeInitIterator(subject,0,REDIS_TAIL); 632 } 633 634 while (listTypeNext(li,&entry)) { 635 if (listTypeEqual(&entry,obj)) { 636 listTypeDelete(&entry); 637 server.dirty++; 638 removed++; 639 if (toremove && removed == toremove) break; 640 } 641 } 642 listTypeReleaseIterator(li); 643 644 /* Clean up raw encoded object */ 645 if (subject->encoding == REDIS_ENCODING_ZIPLIST) 646 decrRefCount(obj); 647 648 if (listTypeLength(subject) == 0) dbDelete(c->db,c->argv[1]); 649 addReplyLongLong(c,removed); 650 if (removed) signalModifiedKey(c->db,c->argv[1]); 651 } 652 653 /* This is the semantic of this command: 654 * RPOPLPUSH srclist dstlist: 655 * IF LLEN(srclist) > 0 656 * element = RPOP srclist 657 * LPUSH dstlist element 658 * RETURN element 659 * ELSE 660 * RETURN nil 661 * END 662 * END 663 * 664 * The idea is to be able to get an element from a list in a reliable way 665 * since the element is not just returned but pushed against another list 666 * as well. This command was originally proposed by Ezra Zygmuntowicz. 667 */ 668 669 void rpoplpushHandlePush(redisClient *origclient, redisClient *c, robj *dstkey, robj *dstobj, robj *value) { 670 if (!handleClientsWaitingListPush(origclient,dstkey,value)) { 671 /* Create the list if the key does not exist */ 672 if (!dstobj) { 673 dstobj = createZiplistObject(); 674 dbAdd(c->db,dstkey,dstobj); 675 } else { 676 signalModifiedKey(c->db,dstkey); 677 } 678 listTypePush(dstobj,value,REDIS_HEAD); 679 /* Additionally propagate this PUSH operation together with 680 * the operation performed by the command. */ 681 { 682 robj **argv = zmalloc(sizeof(robj*)*3); 683 argv[0] = createStringObject("LPUSH",5); 684 argv[1] = dstkey; 685 argv[2] = value; 686 incrRefCount(argv[1]); 687 incrRefCount(argv[2]); 688 alsoPropagate(server.lpushCommand,c->db->id,argv,3, 689 REDIS_PROPAGATE_AOF|REDIS_PROPAGATE_REPL); 690 } 691 } 692 /* Always send the pushed value to the client. */ 693 addReplyBulk(c,value); 694 } 695 696 void rpoplpushCommand(redisClient *c) { 697 robj *sobj, *value; 698 if ((sobj = lookupKeyWriteOrReply(c,c->argv[1],shared.nullbulk)) == NULL || 699 checkType(c,sobj,REDIS_LIST)) return; 700 701 if (listTypeLength(sobj) == 0) { 702 addReply(c,shared.nullbulk); 703 } else { 704 robj *dobj = lookupKeyWrite(c->db,c->argv[2]); 705 robj *touchedkey = c->argv[1]; 706 707 if (dobj && checkType(c,dobj,REDIS_LIST)) return; 708 value = listTypePop(sobj,REDIS_TAIL); 709 /* We saved touched key, and protect it, since rpoplpushHandlePush 710 * may change the client command argument vector. */ 711 incrRefCount(touchedkey); 712 rpoplpushHandlePush(c,c,c->argv[2],dobj,value); 713 714 /* listTypePop returns an object with its refcount incremented */ 715 decrRefCount(value); 716 717 /* Delete the source list when it is empty */ 718 if (listTypeLength(sobj) == 0) dbDelete(c->db,touchedkey); 719 signalModifiedKey(c->db,touchedkey); 720 decrRefCount(touchedkey); 721 server.dirty++; 722 723 /* Replicate this as a simple RPOP since the LPUSH side is replicated 724 * by rpoplpushHandlePush() call if needed (it may not be needed 725 * if a client is blocking wait a push against the list). */ 726 rewriteClientCommandVector(c,2, 727 resetRefCount(createStringObject("RPOP",4)), 728 c->argv[1]); 729 } 730 } 731 732 /*----------------------------------------------------------------------------- 733 * Blocking POP operations 734 *----------------------------------------------------------------------------*/ 735 736 /* Currently Redis blocking operations support is limited to list POP ops, 737 * so the current implementation is not fully generic, but it is also not 738 * completely specific so it will not require a rewrite to support new 739 * kind of blocking operations in the future. 740 * 741 * Still it's important to note that list blocking operations can be already 742 * used as a notification mechanism in order to implement other blocking 743 * operations at application level, so there must be a very strong evidence 744 * of usefulness and generality before new blocking operations are implemented. 745 * 746 * This is how the current blocking POP works, we use BLPOP as example: 747 * - If the user calls BLPOP and the key exists and contains a non empty list 748 * then LPOP is called instead. So BLPOP is semantically the same as LPOP 749 * if there is not to block. 750 * - If instead BLPOP is called and the key does not exists or the list is 751 * empty we need to block. In order to do so we remove the notification for 752 * new data to read in the client socket (so that we'll not serve new 753 * requests if the blocking request is not served). Also we put the client 754 * in a dictionary (db->blocking_keys) mapping keys to a list of clients 755 * blocking for this keys. 756 * - If a PUSH operation against a key with blocked clients waiting is 757 * performed, we serve the first in the list: basically instead to push 758 * the new element inside the list we return it to the (first / oldest) 759 * blocking client, unblock the client, and remove it form the list. 760 * 761 * The above comment and the source code should be enough in order to understand 762 * the implementation and modify / fix it later. 763 */ 764 765 /* Set a client in blocking mode for the specified key, with the specified 766 * timeout */ 767 void blockForKeys(redisClient *c, robj **keys, int numkeys, time_t timeout, robj *target) { 768 dictEntry *de; 769 list *l; 770 int j; 771 772 c->bpop.keys = zmalloc(sizeof(robj*)*numkeys); 773 c->bpop.count = numkeys; 774 c->bpop.timeout = timeout; 775 c->bpop.target = target; 776 777 if (target != NULL) { 778 incrRefCount(target); 779 } 780 781 for (j = 0; j < numkeys; j++) { 782 /* Add the key in the client structure, to map clients -> keys */ 783 c->bpop.keys[j] = keys[j]; 784 incrRefCount(keys[j]); 785 786 /* And in the other "side", to map keys -> clients */ 787 de = dictFind(c->db->blocking_keys,keys[j]); 788 if (de == NULL) { 789 int retval; 790 791 /* For every key we take a list of clients blocked for it */ 792 l = listCreate(); 793 retval = dictAdd(c->db->blocking_keys,keys[j],l); 794 incrRefCount(keys[j]); 795 redisAssertWithInfo(c,keys[j],retval == DICT_OK); 796 } else { 797 l = dictGetVal(de); 798 } 799 listAddNodeTail(l,c); 800 } 801 /* Mark the client as a blocked client */ 802 c->flags |= REDIS_BLOCKED; 803 server.bpop_blocked_clients++; 804 } 805 806 /* Unblock a client that's waiting in a blocking operation such as BLPOP */ 807 void unblockClientWaitingData(redisClient *c) { 808 dictEntry *de; 809 list *l; 810 int j; 811 812 redisAssertWithInfo(c,NULL,c->bpop.keys != NULL); 813 /* The client may wait for multiple keys, so unblock it for every key. */ 814 for (j = 0; j < c->bpop.count; j++) { 815 /* Remove this client from the list of clients waiting for this key. */ 816 de = dictFind(c->db->blocking_keys,c->bpop.keys[j]); 817 redisAssertWithInfo(c,c->bpop.keys[j],de != NULL); 818 l = dictGetVal(de); 819 listDelNode(l,listSearchKey(l,c)); 820 /* If the list is empty we need to remove it to avoid wasting memory */ 821 if (listLength(l) == 0) 822 dictDelete(c->db->blocking_keys,c->bpop.keys[j]); 823 decrRefCount(c->bpop.keys[j]); 824 } 825 826 /* Cleanup the client structure */ 827 zfree(c->bpop.keys); 828 c->bpop.keys = NULL; 829 if (c->bpop.target) decrRefCount(c->bpop.target); 830 c->bpop.target = NULL; 831 c->flags &= ~REDIS_BLOCKED; 832 c->flags |= REDIS_UNBLOCKED; 833 server.bpop_blocked_clients--; 834 listAddNodeTail(server.unblocked_clients,c); 835 } 836 837 /* This should be called from any function PUSHing into lists. 838 * 'c' is the "pushing client", 'key' is the key it is pushing data against, 839 * 'ele' is the element pushed. 840 * 841 * If the function returns 0 there was no client waiting for a list push 842 * against this key. 843 * 844 * If the function returns 1 there was a client waiting for a list push 845 * against this key, the element was passed to this client thus it's not 846 * needed to actually add it to the list and the caller should return asap. */ 847 int handleClientsWaitingListPush(redisClient *c, robj *key, robj *ele) { 848 struct dictEntry *de; 849 redisClient *receiver; 850 int numclients; 851 list *clients; 852 listNode *ln; 853 robj *dstkey, *dstobj; 854 855 de = dictFind(c->db->blocking_keys,key); 856 if (de == NULL) return 0; 857 clients = dictGetVal(de); 858 numclients = listLength(clients); 859 860 /* Try to handle the push as long as there are clients waiting for a push. 861 * Note that "numclients" is used because the list of clients waiting for a 862 * push on "key" is deleted by unblockClient() when empty. 863 * 864 * This loop will have more than 1 iteration when there is a BRPOPLPUSH 865 * that cannot push the target list because it does not contain a list. If 866 * this happens, it simply tries the next client waiting for a push. */ 867 while (numclients--) { 868 ln = listFirst(clients); 869 redisAssertWithInfo(c,key,ln != NULL); 870 receiver = ln->value; 871 dstkey = receiver->bpop.target; 872 873 /* Protect receiver->bpop.target, that will be freed by 874 * the next unblockClientWaitingData() call. */ 875 if (dstkey) incrRefCount(dstkey); 876 877 /* This should remove the first element of the "clients" list. */ 878 unblockClientWaitingData(receiver); 879 880 if (dstkey == NULL) { 881 /* BRPOP/BLPOP */ 882 addReplyMultiBulkLen(receiver,2); 883 addReplyBulk(receiver,key); 884 addReplyBulk(receiver,ele); 885 return 1; /* Serve just the first client as in B[RL]POP semantics */ 886 } else { 887 /* BRPOPLPUSH, note that receiver->db is always equal to c->db. */ 888 dstobj = lookupKeyWrite(receiver->db,dstkey); 889 if (!(dstobj && checkType(receiver,dstobj,REDIS_LIST))) { 890 rpoplpushHandlePush(c,receiver,dstkey,dstobj,ele); 891 decrRefCount(dstkey); 892 return 1; 893 } 894 decrRefCount(dstkey); 895 } 896 } 897 898 return 0; 899 } 900 901 int getTimeoutFromObjectOrReply(redisClient *c, robj *object, time_t *timeout) { 902 long tval; 903 904 if (getLongFromObjectOrReply(c,object,&tval, 905 "timeout is not an integer or out of range") != REDIS_OK) 906 return REDIS_ERR; 907 908 if (tval < 0) { 909 addReplyError(c,"timeout is negative"); 910 return REDIS_ERR; 911 } 912 913 if (tval > 0) tval += server.unixtime; 914 *timeout = tval; 915 916 return REDIS_OK; 917 } 918 919 /* Blocking RPOP/LPOP */ 920 void blockingPopGenericCommand(redisClient *c, int where) { 921 robj *o; 922 time_t timeout; 923 int j; 924 925 if (getTimeoutFromObjectOrReply(c,c->argv[c->argc-1],&timeout) != REDIS_OK) 926 return; 927 928 for (j = 1; j < c->argc-1; j++) { 929 o = lookupKeyWrite(c->db,c->argv[j]); 930 if (o != NULL) { 931 if (o->type != REDIS_LIST) { 932 addReply(c,shared.wrongtypeerr); 933 return; 934 } else { 935 if (listTypeLength(o) != 0) { 936 /* Non empty list, this is like a non normal [LR]POP. */ 937 robj *value = listTypePop(o,where); 938 redisAssert(value != NULL); 939 940 addReplyMultiBulkLen(c,2); 941 addReplyBulk(c,c->argv[j]); 942 addReplyBulk(c,value); 943 decrRefCount(value); 944 if (listTypeLength(o) == 0) dbDelete(c->db,c->argv[j]); 945 signalModifiedKey(c->db,c->argv[j]); 946 server.dirty++; 947 948 /* Replicate it as an [LR]POP instead of B[LR]POP. */ 949 rewriteClientCommandVector(c,2, 950 (where == REDIS_HEAD) ? shared.lpop : shared.rpop, 951 c->argv[j]); 952 return; 953 } 954 } 955 } 956 } 957 958 /* If we are inside a MULTI/EXEC and the list is empty the only thing 959 * we can do is treating it as a timeout (even with timeout 0). */ 960 if (c->flags & REDIS_MULTI) { 961 addReply(c,shared.nullmultibulk); 962 return; 963 } 964 965 /* If the list is empty or the key does not exists we must block */ 966 blockForKeys(c, c->argv + 1, c->argc - 2, timeout, NULL); 967 } 968 969 void blpopCommand(redisClient *c) { 970 blockingPopGenericCommand(c,REDIS_HEAD); 971 } 972 973 void brpopCommand(redisClient *c) { 974 blockingPopGenericCommand(c,REDIS_TAIL); 975 } 976 977 void brpoplpushCommand(redisClient *c) { 978 time_t timeout; 979 980 if (getTimeoutFromObjectOrReply(c,c->argv[3],&timeout) != REDIS_OK) 981 return; 982 983 robj *key = lookupKeyWrite(c->db, c->argv[1]); 984 985 if (key == NULL) { 986 if (c->flags & REDIS_MULTI) { 987 988 /* Blocking against an empty list in a multi state 989 * returns immediately. */ 990 addReply(c, shared.nullbulk); 991 } else { 992 /* The list is empty and the client blocks. */ 993 blockForKeys(c, c->argv + 1, 1, timeout, c->argv[2]); 994 } 995 } else { 996 if (key->type != REDIS_LIST) { 997 addReply(c, shared.wrongtypeerr); 998 } else { 999 1000 /* The list exists and has elements, so 1001 * the regular rpoplpushCommand is executed. */ 1002 redisAssertWithInfo(c,key,listTypeLength(key) > 0); 1003 rpoplpushCommand(c); 1004 } 1005 } 1006 } 1007