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