1 /* 2 * Copyright (c) 2009-2012, Salvatore Sanfilippo <antirez at gmail dot com> 3 * All rights reserved. 4 * 5 * Redistribution and use in source and binary forms, with or without 6 * modification, are permitted provided that the following conditions are met: 7 * 8 * * Redistributions of source code must retain the above copyright notice, 9 * this list of conditions and the following disclaimer. 10 * * Redistributions in binary form must reproduce the above copyright 11 * notice, this list of conditions and the following disclaimer in the 12 * documentation and/or other materials provided with the distribution. 13 * * Neither the name of Redis nor the names of its contributors may be used 14 * to endorse or promote products derived from this software without 15 * specific prior written permission. 16 * 17 * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" 18 * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE 19 * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE 20 * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE 21 * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR 22 * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF 23 * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS 24 * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN 25 * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) 26 * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE 27 * POSSIBILITY OF SUCH DAMAGE. 28 */ 29 30 #include "redis.h" 31 32 /*----------------------------------------------------------------------------- 33 * List API 34 *----------------------------------------------------------------------------*/ 35 36 /* Check the argument length to see if it requires us to convert the ziplist 37 * to a real list. Only check raw-encoded objects because integer encoded 38 * objects are never too long. */ 39 void listTypeTryConversion(robj *subject, robj *value) { 40 if (subject->encoding != REDIS_ENCODING_ZIPLIST) return; 41 if (sdsEncodedObject(value) && 42 sdslen(value->ptr) > server.list_max_ziplist_value) 43 listTypeConvert(subject,REDIS_ENCODING_LINKEDLIST); 44 } 45 46 /* The function pushes an element to the specified list object 'subject', 47 * at head or tail position as specified by 'where'. 48 * 49 * There is no need for the caller to increment the refcount of 'value' as 50 * the function takes care of it if needed. */ 51 void listTypePush(robj *subject, robj *value, int where) { 52 /* Check if we need to convert the ziplist */ 53 listTypeTryConversion(subject,value); 54 if (subject->encoding == REDIS_ENCODING_ZIPLIST && 55 ziplistLen(subject->ptr) >= server.list_max_ziplist_entries) 56 listTypeConvert(subject,REDIS_ENCODING_LINKEDLIST); 57 58 if (subject->encoding == REDIS_ENCODING_ZIPLIST) { 59 int pos = (where == REDIS_HEAD) ? ZIPLIST_HEAD : ZIPLIST_TAIL; 60 value = getDecodedObject(value); 61 subject->ptr = ziplistPush(subject->ptr,value->ptr,sdslen(value->ptr),pos); 62 decrRefCount(value); 63 } else if (subject->encoding == REDIS_ENCODING_LINKEDLIST) { 64 if (where == REDIS_HEAD) { 65 listAddNodeHead(subject->ptr,value); 66 } else { 67 listAddNodeTail(subject->ptr,value); 68 } 69 incrRefCount(value); 70 } else { 71 redisPanic("Unknown list encoding"); 72 } 73 } 74 75 robj *listTypePop(robj *subject, int where) { 76 robj *value = NULL; 77 if (subject->encoding == REDIS_ENCODING_ZIPLIST) { 78 unsigned char *p; 79 unsigned char *vstr; 80 unsigned int vlen; 81 long long vlong; 82 int pos = (where == REDIS_HEAD) ? 0 : -1; 83 p = ziplistIndex(subject->ptr,pos); 84 if (ziplistGet(p,&vstr,&vlen,&vlong)) { 85 if (vstr) { 86 value = createStringObject((char*)vstr,vlen); 87 } else { 88 value = createStringObjectFromLongLong(vlong); 89 } 90 /* We only need to delete an element when it exists */ 91 subject->ptr = ziplistDelete(subject->ptr,&p); 92 } 93 } else if (subject->encoding == REDIS_ENCODING_LINKEDLIST) { 94 list *list = subject->ptr; 95 listNode *ln; 96 if (where == REDIS_HEAD) { 97 ln = listFirst(list); 98 } else { 99 ln = listLast(list); 100 } 101 if (ln != NULL) { 102 value = listNodeValue(ln); 103 incrRefCount(value); 104 listDelNode(list,ln); 105 } 106 } else { 107 redisPanic("Unknown list encoding"); 108 } 109 return value; 110 } 111 112 unsigned long listTypeLength(robj *subject) { 113 if (subject->encoding == REDIS_ENCODING_ZIPLIST) { 114 return ziplistLen(subject->ptr); 115 } else if (subject->encoding == REDIS_ENCODING_LINKEDLIST) { 116 return listLength((list*)subject->ptr); 117 } else { 118 redisPanic("Unknown list encoding"); 119 } 120 } 121 122 /* Initialize an iterator at the specified index. */ 123 listTypeIterator *listTypeInitIterator(robj *subject, long index, unsigned char direction) { 124 listTypeIterator *li = zmalloc(sizeof(listTypeIterator)); 125 li->subject = subject; 126 li->encoding = subject->encoding; 127 li->direction = direction; 128 if (li->encoding == REDIS_ENCODING_ZIPLIST) { 129 li->zi = ziplistIndex(subject->ptr,index); 130 } else if (li->encoding == REDIS_ENCODING_LINKEDLIST) { 131 li->ln = listIndex(subject->ptr,index); 132 } else { 133 redisPanic("Unknown list encoding"); 134 } 135 return li; 136 } 137 138 /* Clean up the iterator. */ 139 void listTypeReleaseIterator(listTypeIterator *li) { 140 zfree(li); 141 } 142 143 /* Stores pointer to current the entry in the provided entry structure 144 * and advances the position of the iterator. Returns 1 when the current 145 * entry is in fact an entry, 0 otherwise. */ 146 int listTypeNext(listTypeIterator *li, listTypeEntry *entry) { 147 /* Protect from converting when iterating */ 148 redisAssert(li->subject->encoding == li->encoding); 149 150 entry->li = li; 151 if (li->encoding == REDIS_ENCODING_ZIPLIST) { 152 entry->zi = li->zi; 153 if (entry->zi != NULL) { 154 if (li->direction == REDIS_TAIL) 155 li->zi = ziplistNext(li->subject->ptr,li->zi); 156 else 157 li->zi = ziplistPrev(li->subject->ptr,li->zi); 158 return 1; 159 } 160 } else if (li->encoding == REDIS_ENCODING_LINKEDLIST) { 161 entry->ln = li->ln; 162 if (entry->ln != NULL) { 163 if (li->direction == REDIS_TAIL) 164 li->ln = li->ln->next; 165 else 166 li->ln = li->ln->prev; 167 return 1; 168 } 169 } else { 170 redisPanic("Unknown list encoding"); 171 } 172 return 0; 173 } 174 175 /* Return entry or NULL at the current position of the iterator. */ 176 robj *listTypeGet(listTypeEntry *entry) { 177 listTypeIterator *li = entry->li; 178 robj *value = NULL; 179 if (li->encoding == REDIS_ENCODING_ZIPLIST) { 180 unsigned char *vstr; 181 unsigned int vlen; 182 long long vlong; 183 redisAssert(entry->zi != NULL); 184 if (ziplistGet(entry->zi,&vstr,&vlen,&vlong)) { 185 if (vstr) { 186 value = createStringObject((char*)vstr,vlen); 187 } else { 188 value = createStringObjectFromLongLong(vlong); 189 } 190 } 191 } else if (li->encoding == REDIS_ENCODING_LINKEDLIST) { 192 redisAssert(entry->ln != NULL); 193 value = listNodeValue(entry->ln); 194 incrRefCount(value); 195 } else { 196 redisPanic("Unknown list encoding"); 197 } 198 return value; 199 } 200 201 void listTypeInsert(listTypeEntry *entry, robj *value, int where) { 202 robj *subject = entry->li->subject; 203 if (entry->li->encoding == REDIS_ENCODING_ZIPLIST) { 204 value = getDecodedObject(value); 205 if (where == REDIS_TAIL) { 206 unsigned char *next = ziplistNext(subject->ptr,entry->zi); 207 208 /* When we insert after the current element, but the current element 209 * is the tail of the list, we need to do a push. */ 210 if (next == NULL) { 211 subject->ptr = ziplistPush(subject->ptr,value->ptr,sdslen(value->ptr),REDIS_TAIL); 212 } else { 213 subject->ptr = ziplistInsert(subject->ptr,next,value->ptr,sdslen(value->ptr)); 214 } 215 } else { 216 subject->ptr = ziplistInsert(subject->ptr,entry->zi,value->ptr,sdslen(value->ptr)); 217 } 218 decrRefCount(value); 219 } else if (entry->li->encoding == REDIS_ENCODING_LINKEDLIST) { 220 if (where == REDIS_TAIL) { 221 listInsertNode(subject->ptr,entry->ln,value,AL_START_TAIL); 222 } else { 223 listInsertNode(subject->ptr,entry->ln,value,AL_START_HEAD); 224 } 225 incrRefCount(value); 226 } else { 227 redisPanic("Unknown list encoding"); 228 } 229 } 230 231 /* Compare the given object with the entry at the current position. */ 232 int listTypeEqual(listTypeEntry *entry, robj *o) { 233 listTypeIterator *li = entry->li; 234 if (li->encoding == REDIS_ENCODING_ZIPLIST) { 235 redisAssertWithInfo(NULL,o,sdsEncodedObject(o)); 236 return ziplistCompare(entry->zi,o->ptr,sdslen(o->ptr)); 237 } else if (li->encoding == REDIS_ENCODING_LINKEDLIST) { 238 return equalStringObjects(o,listNodeValue(entry->ln)); 239 } else { 240 redisPanic("Unknown list encoding"); 241 } 242 } 243 244 /* Delete the element pointed to. */ 245 void listTypeDelete(listTypeEntry *entry) { 246 listTypeIterator *li = entry->li; 247 if (li->encoding == REDIS_ENCODING_ZIPLIST) { 248 unsigned char *p = entry->zi; 249 li->subject->ptr = ziplistDelete(li->subject->ptr,&p); 250 251 /* Update position of the iterator depending on the direction */ 252 if (li->direction == REDIS_TAIL) 253 li->zi = p; 254 else 255 li->zi = ziplistPrev(li->subject->ptr,p); 256 } else if (entry->li->encoding == REDIS_ENCODING_LINKEDLIST) { 257 listNode *next; 258 if (li->direction == REDIS_TAIL) 259 next = entry->ln->next; 260 else 261 next = entry->ln->prev; 262 listDelNode(li->subject->ptr,entry->ln); 263 li->ln = next; 264 } else { 265 redisPanic("Unknown list encoding"); 266 } 267 } 268 269 void listTypeConvert(robj *subject, int enc) { 270 listTypeIterator *li; 271 listTypeEntry entry; 272 redisAssertWithInfo(NULL,subject,subject->type == REDIS_LIST); 273 274 if (enc == REDIS_ENCODING_LINKEDLIST) { 275 list *l = listCreate(); 276 listSetFreeMethod(l,decrRefCountVoid); 277 278 /* listTypeGet returns a robj with incremented refcount */ 279 li = listTypeInitIterator(subject,0,REDIS_TAIL); 280 while (listTypeNext(li,&entry)) listAddNodeTail(l,listTypeGet(&entry)); 281 listTypeReleaseIterator(li); 282 283 subject->encoding = REDIS_ENCODING_LINKEDLIST; 284 zfree(subject->ptr); 285 subject->ptr = l; 286 } else { 287 redisPanic("Unsupported list conversion"); 288 } 289 } 290 291 /*----------------------------------------------------------------------------- 292 * List Commands 293 *----------------------------------------------------------------------------*/ 294 295 void pushGenericCommand(redisClient *c, int where) { 296 int j, waiting = 0, pushed = 0; 297 robj *lobj = lookupKeyWrite(c->db,c->argv[1]); 298 299 if (lobj && lobj->type != REDIS_LIST) { 300 addReply(c,shared.wrongtypeerr); 301 return; 302 } 303 304 for (j = 2; j < c->argc; j++) { 305 c->argv[j] = tryObjectEncoding(c->argv[j]); 306 if (!lobj) { 307 lobj = createZiplistObject(); 308 dbAdd(c->db,c->argv[1],lobj); 309 } 310 listTypePush(lobj,c->argv[j],where); 311 pushed++; 312 } 313 addReplyLongLong(c, waiting + (lobj ? listTypeLength(lobj) : 0)); 314 if (pushed) { 315 char *event = (where == REDIS_HEAD) ? "lpush" : "rpush"; 316 317 signalModifiedKey(c->db,c->argv[1]); 318 notifyKeyspaceEvent(REDIS_NOTIFY_LIST,event,c->argv[1],c->db->id); 319 } 320 server.dirty += pushed; 321 } 322 323 void lpushCommand(redisClient *c) { 324 pushGenericCommand(c,REDIS_HEAD); 325 } 326 327 void rpushCommand(redisClient *c) { 328 pushGenericCommand(c,REDIS_TAIL); 329 } 330 331 void pushxGenericCommand(redisClient *c, robj *refval, robj *val, int where) { 332 robj *subject; 333 listTypeIterator *iter; 334 listTypeEntry entry; 335 int inserted = 0; 336 337 if ((subject = lookupKeyReadOrReply(c,c->argv[1],shared.czero)) == NULL || 338 checkType(c,subject,REDIS_LIST)) return; 339 340 if (refval != NULL) { 341 /* We're not sure if this value can be inserted yet, but we cannot 342 * convert the list inside the iterator. We don't want to loop over 343 * the list twice (once to see if the value can be inserted and once 344 * to do the actual insert), so we assume this value can be inserted 345 * and convert the ziplist to a regular list if necessary. */ 346 listTypeTryConversion(subject,val); 347 348 /* Seek refval from head to tail */ 349 iter = listTypeInitIterator(subject,0,REDIS_TAIL); 350 while (listTypeNext(iter,&entry)) { 351 if (listTypeEqual(&entry,refval)) { 352 listTypeInsert(&entry,val,where); 353 inserted = 1; 354 break; 355 } 356 } 357 listTypeReleaseIterator(iter); 358 359 if (inserted) { 360 /* Check if the length exceeds the ziplist length threshold. */ 361 if (subject->encoding == REDIS_ENCODING_ZIPLIST && 362 ziplistLen(subject->ptr) > server.list_max_ziplist_entries) 363 listTypeConvert(subject,REDIS_ENCODING_LINKEDLIST); 364 signalModifiedKey(c->db,c->argv[1]); 365 notifyKeyspaceEvent(REDIS_NOTIFY_LIST,"linsert", 366 c->argv[1],c->db->id); 367 server.dirty++; 368 } else { 369 /* Notify client of a failed insert */ 370 addReply(c,shared.cnegone); 371 return; 372 } 373 } else { 374 char *event = (where == REDIS_HEAD) ? "lpush" : "rpush"; 375 376 listTypePush(subject,val,where); 377 signalModifiedKey(c->db,c->argv[1]); 378 notifyKeyspaceEvent(REDIS_NOTIFY_LIST,event,c->argv[1],c->db->id); 379 server.dirty++; 380 } 381 382 addReplyLongLong(c,listTypeLength(subject)); 383 } 384 385 void lpushxCommand(redisClient *c) { 386 c->argv[2] = tryObjectEncoding(c->argv[2]); 387 pushxGenericCommand(c,NULL,c->argv[2],REDIS_HEAD); 388 } 389 390 void rpushxCommand(redisClient *c) { 391 c->argv[2] = tryObjectEncoding(c->argv[2]); 392 pushxGenericCommand(c,NULL,c->argv[2],REDIS_TAIL); 393 } 394 395 void linsertCommand(redisClient *c) { 396 c->argv[4] = tryObjectEncoding(c->argv[4]); 397 if (strcasecmp(c->argv[2]->ptr,"after") == 0) { 398 pushxGenericCommand(c,c->argv[3],c->argv[4],REDIS_TAIL); 399 } else if (strcasecmp(c->argv[2]->ptr,"before") == 0) { 400 pushxGenericCommand(c,c->argv[3],c->argv[4],REDIS_HEAD); 401 } else { 402 addReply(c,shared.syntaxerr); 403 } 404 } 405 406 void llenCommand(redisClient *c) { 407 robj *o = lookupKeyReadOrReply(c,c->argv[1],shared.czero); 408 if (o == NULL || checkType(c,o,REDIS_LIST)) return; 409 addReplyLongLong(c,listTypeLength(o)); 410 } 411 412 void lindexCommand(redisClient *c) { 413 robj *o = lookupKeyReadOrReply(c,c->argv[1],shared.nullbulk); 414 if (o == NULL || checkType(c,o,REDIS_LIST)) return; 415 long index; 416 robj *value = NULL; 417 418 if ((getLongFromObjectOrReply(c, c->argv[2], &index, NULL) != REDIS_OK)) 419 return; 420 421 if (o->encoding == REDIS_ENCODING_ZIPLIST) { 422 unsigned char *p; 423 unsigned char *vstr; 424 unsigned int vlen; 425 long long vlong; 426 p = ziplistIndex(o->ptr,index); 427 if (ziplistGet(p,&vstr,&vlen,&vlong)) { 428 if (vstr) { 429 value = createStringObject((char*)vstr,vlen); 430 } else { 431 value = createStringObjectFromLongLong(vlong); 432 } 433 addReplyBulk(c,value); 434 decrRefCount(value); 435 } else { 436 addReply(c,shared.nullbulk); 437 } 438 } else if (o->encoding == REDIS_ENCODING_LINKEDLIST) { 439 listNode *ln = listIndex(o->ptr,index); 440 if (ln != NULL) { 441 value = listNodeValue(ln); 442 addReplyBulk(c,value); 443 } else { 444 addReply(c,shared.nullbulk); 445 } 446 } else { 447 redisPanic("Unknown list encoding"); 448 } 449 } 450 451 void lsetCommand(redisClient *c) { 452 robj *o = lookupKeyWriteOrReply(c,c->argv[1],shared.nokeyerr); 453 if (o == NULL || checkType(c,o,REDIS_LIST)) return; 454 long index; 455 robj *value = (c->argv[3] = tryObjectEncoding(c->argv[3])); 456 457 if ((getLongFromObjectOrReply(c, c->argv[2], &index, NULL) != REDIS_OK)) 458 return; 459 460 listTypeTryConversion(o,value); 461 if (o->encoding == REDIS_ENCODING_ZIPLIST) { 462 unsigned char *p, *zl = o->ptr; 463 p = ziplistIndex(zl,index); 464 if (p == NULL) { 465 addReply(c,shared.outofrangeerr); 466 } else { 467 o->ptr = ziplistDelete(o->ptr,&p); 468 value = getDecodedObject(value); 469 o->ptr = ziplistInsert(o->ptr,p,value->ptr,sdslen(value->ptr)); 470 decrRefCount(value); 471 addReply(c,shared.ok); 472 signalModifiedKey(c->db,c->argv[1]); 473 notifyKeyspaceEvent(REDIS_NOTIFY_LIST,"lset",c->argv[1],c->db->id); 474 server.dirty++; 475 } 476 } else if (o->encoding == REDIS_ENCODING_LINKEDLIST) { 477 listNode *ln = listIndex(o->ptr,index); 478 if (ln == NULL) { 479 addReply(c,shared.outofrangeerr); 480 } else { 481 decrRefCount((robj*)listNodeValue(ln)); 482 listNodeValue(ln) = value; 483 incrRefCount(value); 484 addReply(c,shared.ok); 485 signalModifiedKey(c->db,c->argv[1]); 486 notifyKeyspaceEvent(REDIS_NOTIFY_LIST,"lset",c->argv[1],c->db->id); 487 server.dirty++; 488 } 489 } else { 490 redisPanic("Unknown list encoding"); 491 } 492 } 493 494 void popGenericCommand(redisClient *c, int where) { 495 robj *o = lookupKeyWriteOrReply(c,c->argv[1],shared.nullbulk); 496 if (o == NULL || checkType(c,o,REDIS_LIST)) return; 497 498 robj *value = listTypePop(o,where); 499 if (value == NULL) { 500 addReply(c,shared.nullbulk); 501 } else { 502 char *event = (where == REDIS_HEAD) ? "lpop" : "rpop"; 503 504 addReplyBulk(c,value); 505 decrRefCount(value); 506 notifyKeyspaceEvent(REDIS_NOTIFY_LIST,event,c->argv[1],c->db->id); 507 if (listTypeLength(o) == 0) { 508 notifyKeyspaceEvent(REDIS_NOTIFY_GENERIC,"del", 509 c->argv[1],c->db->id); 510 dbDelete(c->db,c->argv[1]); 511 } 512 signalModifiedKey(c->db,c->argv[1]); 513 server.dirty++; 514 } 515 } 516 517 void lpopCommand(redisClient *c) { 518 popGenericCommand(c,REDIS_HEAD); 519 } 520 521 void rpopCommand(redisClient *c) { 522 popGenericCommand(c,REDIS_TAIL); 523 } 524 525 void lrangeCommand(redisClient *c) { 526 robj *o; 527 long start, end, llen, rangelen; 528 529 if ((getLongFromObjectOrReply(c, c->argv[2], &start, NULL) != REDIS_OK) || 530 (getLongFromObjectOrReply(c, c->argv[3], &end, NULL) != REDIS_OK)) return; 531 532 if ((o = lookupKeyReadOrReply(c,c->argv[1],shared.emptymultibulk)) == NULL 533 || checkType(c,o,REDIS_LIST)) return; 534 llen = listTypeLength(o); 535 536 /* convert negative indexes */ 537 if (start < 0) start = llen+start; 538 if (end < 0) end = llen+end; 539 if (start < 0) start = 0; 540 541 /* Invariant: start >= 0, so this test will be true when end < 0. 542 * The range is empty when start > end or start >= length. */ 543 if (start > end || start >= llen) { 544 addReply(c,shared.emptymultibulk); 545 return; 546 } 547 if (end >= llen) end = llen-1; 548 rangelen = (end-start)+1; 549 550 /* Return the result in form of a multi-bulk reply */ 551 addReplyMultiBulkLen(c,rangelen); 552 if (o->encoding == REDIS_ENCODING_ZIPLIST) { 553 unsigned char *p = ziplistIndex(o->ptr,start); 554 unsigned char *vstr; 555 unsigned int vlen; 556 long long vlong; 557 558 while(rangelen--) { 559 ziplistGet(p,&vstr,&vlen,&vlong); 560 if (vstr) { 561 addReplyBulkCBuffer(c,vstr,vlen); 562 } else { 563 addReplyBulkLongLong(c,vlong); 564 } 565 p = ziplistNext(o->ptr,p); 566 } 567 } else if (o->encoding == REDIS_ENCODING_LINKEDLIST) { 568 listNode *ln; 569 570 /* If we are nearest to the end of the list, reach the element 571 * starting from tail and going backward, as it is faster. */ 572 if (start > llen/2) start -= llen; 573 ln = listIndex(o->ptr,start); 574 575 while(rangelen--) { 576 addReplyBulk(c,ln->value); 577 ln = ln->next; 578 } 579 } else { 580 redisPanic("List encoding is not LINKEDLIST nor ZIPLIST!"); 581 } 582 } 583 584 void ltrimCommand(redisClient *c) { 585 robj *o; 586 long start, end, llen, j, ltrim, rtrim; 587 list *list; 588 listNode *ln; 589 590 if ((getLongFromObjectOrReply(c, c->argv[2], &start, NULL) != REDIS_OK) || 591 (getLongFromObjectOrReply(c, c->argv[3], &end, NULL) != REDIS_OK)) return; 592 593 if ((o = lookupKeyWriteOrReply(c,c->argv[1],shared.ok)) == NULL || 594 checkType(c,o,REDIS_LIST)) return; 595 llen = listTypeLength(o); 596 597 /* convert negative indexes */ 598 if (start < 0) start = llen+start; 599 if (end < 0) end = llen+end; 600 if (start < 0) start = 0; 601 602 /* Invariant: start >= 0, so this test will be true when end < 0. 603 * The range is empty when start > end or start >= length. */ 604 if (start > end || start >= llen) { 605 /* Out of range start or start > end result in empty list */ 606 ltrim = llen; 607 rtrim = 0; 608 } else { 609 if (end >= llen) end = llen-1; 610 ltrim = start; 611 rtrim = llen-end-1; 612 } 613 614 /* Remove list elements to perform the trim */ 615 if (o->encoding == REDIS_ENCODING_ZIPLIST) { 616 o->ptr = ziplistDeleteRange(o->ptr,0,ltrim); 617 o->ptr = ziplistDeleteRange(o->ptr,-rtrim,rtrim); 618 } else if (o->encoding == REDIS_ENCODING_LINKEDLIST) { 619 list = o->ptr; 620 for (j = 0; j < ltrim; j++) { 621 ln = listFirst(list); 622 listDelNode(list,ln); 623 } 624 for (j = 0; j < rtrim; j++) { 625 ln = listLast(list); 626 listDelNode(list,ln); 627 } 628 } else { 629 redisPanic("Unknown list encoding"); 630 } 631 632 notifyKeyspaceEvent(REDIS_NOTIFY_LIST,"ltrim",c->argv[1],c->db->id); 633 if (listTypeLength(o) == 0) { 634 dbDelete(c->db,c->argv[1]); 635 notifyKeyspaceEvent(REDIS_NOTIFY_GENERIC,"del",c->argv[1],c->db->id); 636 } 637 signalModifiedKey(c->db,c->argv[1]); 638 server.dirty++; 639 addReply(c,shared.ok); 640 } 641 642 void lremCommand(redisClient *c) { 643 robj *subject, *obj; 644 obj = c->argv[3] = tryObjectEncoding(c->argv[3]); 645 long toremove; 646 long removed = 0; 647 listTypeEntry entry; 648 649 if ((getLongFromObjectOrReply(c, c->argv[2], &toremove, NULL) != REDIS_OK)) 650 return; 651 652 subject = lookupKeyWriteOrReply(c,c->argv[1],shared.czero); 653 if (subject == NULL || checkType(c,subject,REDIS_LIST)) return; 654 655 /* Make sure obj is raw when we're dealing with a ziplist */ 656 if (subject->encoding == REDIS_ENCODING_ZIPLIST) 657 obj = getDecodedObject(obj); 658 659 listTypeIterator *li; 660 if (toremove < 0) { 661 toremove = -toremove; 662 li = listTypeInitIterator(subject,-1,REDIS_HEAD); 663 } else { 664 li = listTypeInitIterator(subject,0,REDIS_TAIL); 665 } 666 667 while (listTypeNext(li,&entry)) { 668 if (listTypeEqual(&entry,obj)) { 669 listTypeDelete(&entry); 670 server.dirty++; 671 removed++; 672 if (toremove && removed == toremove) break; 673 } 674 } 675 listTypeReleaseIterator(li); 676 677 /* Clean up raw encoded object */ 678 if (subject->encoding == REDIS_ENCODING_ZIPLIST) 679 decrRefCount(obj); 680 681 if (listTypeLength(subject) == 0) dbDelete(c->db,c->argv[1]); 682 addReplyLongLong(c,removed); 683 if (removed) signalModifiedKey(c->db,c->argv[1]); 684 } 685 686 /* This is the semantic of this command: 687 * RPOPLPUSH srclist dstlist: 688 * IF LLEN(srclist) > 0 689 * element = RPOP srclist 690 * LPUSH dstlist element 691 * RETURN element 692 * ELSE 693 * RETURN nil 694 * END 695 * END 696 * 697 * The idea is to be able to get an element from a list in a reliable way 698 * since the element is not just returned but pushed against another list 699 * as well. This command was originally proposed by Ezra Zygmuntowicz. 700 */ 701 702 void rpoplpushHandlePush(redisClient *c, robj *dstkey, robj *dstobj, robj *value) { 703 /* Create the list if the key does not exist */ 704 if (!dstobj) { 705 dstobj = createZiplistObject(); 706 dbAdd(c->db,dstkey,dstobj); 707 } 708 signalModifiedKey(c->db,dstkey); 709 listTypePush(dstobj,value,REDIS_HEAD); 710 notifyKeyspaceEvent(REDIS_NOTIFY_LIST,"lpush",dstkey,c->db->id); 711 /* Always send the pushed value to the client. */ 712 addReplyBulk(c,value); 713 } 714 715 void rpoplpushCommand(redisClient *c) { 716 robj *sobj, *value; 717 if ((sobj = lookupKeyWriteOrReply(c,c->argv[1],shared.nullbulk)) == NULL || 718 checkType(c,sobj,REDIS_LIST)) return; 719 720 if (listTypeLength(sobj) == 0) { 721 /* This may only happen after loading very old RDB files. Recent 722 * versions of Redis delete keys of empty lists. */ 723 addReply(c,shared.nullbulk); 724 } else { 725 robj *dobj = lookupKeyWrite(c->db,c->argv[2]); 726 robj *touchedkey = c->argv[1]; 727 728 if (dobj && checkType(c,dobj,REDIS_LIST)) return; 729 value = listTypePop(sobj,REDIS_TAIL); 730 /* We saved touched key, and protect it, since rpoplpushHandlePush 731 * may change the client command argument vector (it does not 732 * currently). */ 733 incrRefCount(touchedkey); 734 rpoplpushHandlePush(c,c->argv[2],dobj,value); 735 736 /* listTypePop returns an object with its refcount incremented */ 737 decrRefCount(value); 738 739 /* Delete the source list when it is empty */ 740 notifyKeyspaceEvent(REDIS_NOTIFY_LIST,"rpop",touchedkey,c->db->id); 741 if (listTypeLength(sobj) == 0) { 742 dbDelete(c->db,touchedkey); 743 notifyKeyspaceEvent(REDIS_NOTIFY_GENERIC,"del", 744 touchedkey,c->db->id); 745 } 746 signalModifiedKey(c->db,touchedkey); 747 decrRefCount(touchedkey); 748 server.dirty++; 749 } 750 } 751 752 /*----------------------------------------------------------------------------- 753 * Blocking POP operations 754 *----------------------------------------------------------------------------*/ 755 756 /* This is how the current blocking POP works, we use BLPOP as example: 757 * - If the user calls BLPOP and the key exists and contains a non empty list 758 * then LPOP is called instead. So BLPOP is semantically the same as LPOP 759 * if blocking is not required. 760 * - If instead BLPOP is called and the key does not exists or the list is 761 * empty we need to block. In order to do so we remove the notification for 762 * new data to read in the client socket (so that we'll not serve new 763 * requests if the blocking request is not served). Also we put the client 764 * in a dictionary (db->blocking_keys) mapping keys to a list of clients 765 * blocking for this keys. 766 * - If a PUSH operation against a key with blocked clients waiting is 767 * performed, we mark this key as "ready", and after the current command, 768 * MULTI/EXEC block, or script, is executed, we serve all the clients waiting 769 * for this list, from the one that blocked first, to the last, accordingly 770 * to the number of elements we have in the ready list. 771 */ 772 773 /* Set a client in blocking mode for the specified key, with the specified 774 * timeout */ 775 void blockForKeys(redisClient *c, robj **keys, int numkeys, mstime_t timeout, robj *target) { 776 dictEntry *de; 777 list *l; 778 int j; 779 780 c->bpop.timeout = timeout; 781 c->bpop.target = target; 782 783 if (target != NULL) incrRefCount(target); 784 785 for (j = 0; j < numkeys; j++) { 786 /* If the key already exists in the dict ignore it. */ 787 if (dictAdd(c->bpop.keys,keys[j],NULL) != DICT_OK) continue; 788 incrRefCount(keys[j]); 789 790 /* And in the other "side", to map keys -> clients */ 791 de = dictFind(c->db->blocking_keys,keys[j]); 792 if (de == NULL) { 793 int retval; 794 795 /* For every key we take a list of clients blocked for it */ 796 l = listCreate(); 797 retval = dictAdd(c->db->blocking_keys,keys[j],l); 798 incrRefCount(keys[j]); 799 redisAssertWithInfo(c,keys[j],retval == DICT_OK); 800 } else { 801 l = dictGetVal(de); 802 } 803 listAddNodeTail(l,c); 804 } 805 blockClient(c,REDIS_BLOCKED_LIST); 806 } 807 808 /* Unblock a client that's waiting in a blocking operation such as BLPOP. 809 * You should never call this function directly, but unblockClient() instead. */ 810 void unblockClientWaitingData(redisClient *c) { 811 dictEntry *de; 812 dictIterator *di; 813 list *l; 814 815 redisAssertWithInfo(c,NULL,dictSize(c->bpop.keys) != 0); 816 di = dictGetIterator(c->bpop.keys); 817 /* The client may wait for multiple keys, so unblock it for every key. */ 818 while((de = dictNext(di)) != NULL) { 819 robj *key = dictGetKey(de); 820 821 /* Remove this client from the list of clients waiting for this key. */ 822 l = dictFetchValue(c->db->blocking_keys,key); 823 redisAssertWithInfo(c,key,l != NULL); 824 listDelNode(l,listSearchKey(l,c)); 825 /* If the list is empty we need to remove it to avoid wasting memory */ 826 if (listLength(l) == 0) 827 dictDelete(c->db->blocking_keys,key); 828 } 829 dictReleaseIterator(di); 830 831 /* Cleanup the client structure */ 832 dictEmpty(c->bpop.keys,NULL); 833 if (c->bpop.target) { 834 decrRefCount(c->bpop.target); 835 c->bpop.target = NULL; 836 } 837 } 838 839 /* If the specified key has clients blocked waiting for list pushes, this 840 * function will put the key reference into the server.ready_keys list. 841 * Note that db->ready_keys is a hash table that allows us to avoid putting 842 * the same key again and again in the list in case of multiple pushes 843 * made by a script or in the context of MULTI/EXEC. 844 * 845 * The list will be finally processed by handleClientsBlockedOnLists() */ 846 void signalListAsReady(redisDb *db, robj *key) { 847 readyList *rl; 848 849 /* No clients blocking for this key? No need to queue it. */ 850 if (dictFind(db->blocking_keys,key) == NULL) return; 851 852 /* Key was already signaled? No need to queue it again. */ 853 if (dictFind(db->ready_keys,key) != NULL) return; 854 855 /* Ok, we need to queue this key into server.ready_keys. */ 856 rl = zmalloc(sizeof(*rl)); 857 rl->key = key; 858 rl->db = db; 859 incrRefCount(key); 860 listAddNodeTail(server.ready_keys,rl); 861 862 /* We also add the key in the db->ready_keys dictionary in order 863 * to avoid adding it multiple times into a list with a simple O(1) 864 * check. */ 865 incrRefCount(key); 866 redisAssert(dictAdd(db->ready_keys,key,NULL) == DICT_OK); 867 } 868 869 /* This is a helper function for handleClientsBlockedOnLists(). It's work 870 * is to serve a specific client (receiver) that is blocked on 'key' 871 * in the context of the specified 'db', doing the following: 872 * 873 * 1) Provide the client with the 'value' element. 874 * 2) If the dstkey is not NULL (we are serving a BRPOPLPUSH) also push the 875 * 'value' element on the destination list (the LPUSH side of the command). 876 * 3) Propagate the resulting BRPOP, BLPOP and additional LPUSH if any into 877 * the AOF and replication channel. 878 * 879 * The argument 'where' is REDIS_TAIL or REDIS_HEAD, and indicates if the 880 * 'value' element was popped fron the head (BLPOP) or tail (BRPOP) so that 881 * we can propagate the command properly. 882 * 883 * The function returns REDIS_OK if we are able to serve the client, otherwise 884 * REDIS_ERR is returned to signal the caller that the list POP operation 885 * should be undone as the client was not served: This only happens for 886 * BRPOPLPUSH that fails to push the value to the destination key as it is 887 * of the wrong type. */ 888 int serveClientBlockedOnList(redisClient *receiver, robj *key, robj *dstkey, redisDb *db, robj *value, int where) 889 { 890 robj *argv[3]; 891 892 if (dstkey == NULL) { 893 /* Propagate the [LR]POP operation. */ 894 argv[0] = (where == REDIS_HEAD) ? shared.lpop : 895 shared.rpop; 896 argv[1] = key; 897 propagate((where == REDIS_HEAD) ? 898 server.lpopCommand : server.rpopCommand, 899 db->id,argv,2,REDIS_PROPAGATE_AOF|REDIS_PROPAGATE_REPL); 900 901 /* BRPOP/BLPOP */ 902 addReplyMultiBulkLen(receiver,2); 903 addReplyBulk(receiver,key); 904 addReplyBulk(receiver,value); 905 } else { 906 /* BRPOPLPUSH */ 907 robj *dstobj = 908 lookupKeyWrite(receiver->db,dstkey); 909 if (!(dstobj && 910 checkType(receiver,dstobj,REDIS_LIST))) 911 { 912 /* Propagate the RPOP operation. */ 913 argv[0] = shared.rpop; 914 argv[1] = key; 915 propagate(server.rpopCommand, 916 db->id,argv,2, 917 REDIS_PROPAGATE_AOF| 918 REDIS_PROPAGATE_REPL); 919 rpoplpushHandlePush(receiver,dstkey,dstobj, 920 value); 921 /* Propagate the LPUSH operation. */ 922 argv[0] = shared.lpush; 923 argv[1] = dstkey; 924 argv[2] = value; 925 propagate(server.lpushCommand, 926 db->id,argv,3, 927 REDIS_PROPAGATE_AOF| 928 REDIS_PROPAGATE_REPL); 929 } else { 930 /* BRPOPLPUSH failed because of wrong 931 * destination type. */ 932 return REDIS_ERR; 933 } 934 } 935 return REDIS_OK; 936 } 937 938 /* This function should be called by Redis every time a single command, 939 * a MULTI/EXEC block, or a Lua script, terminated its execution after 940 * being called by a client. 941 * 942 * All the keys with at least one client blocked that received at least 943 * one new element via some PUSH operation are accumulated into 944 * the server.ready_keys list. This function will run the list and will 945 * serve clients accordingly. Note that the function will iterate again and 946 * again as a result of serving BRPOPLPUSH we can have new blocking clients 947 * to serve because of the PUSH side of BRPOPLPUSH. */ 948 void handleClientsBlockedOnLists(void) { 949 while(listLength(server.ready_keys) != 0) { 950 list *l; 951 952 /* Point server.ready_keys to a fresh list and save the current one 953 * locally. This way as we run the old list we are free to call 954 * signalListAsReady() that may push new elements in server.ready_keys 955 * when handling clients blocked into BRPOPLPUSH. */ 956 l = server.ready_keys; 957 server.ready_keys = listCreate(); 958 959 while(listLength(l) != 0) { 960 listNode *ln = listFirst(l); 961 readyList *rl = ln->value; 962 963 /* First of all remove this key from db->ready_keys so that 964 * we can safely call signalListAsReady() against this key. */ 965 dictDelete(rl->db->ready_keys,rl->key); 966 967 /* If the key exists and it's a list, serve blocked clients 968 * with data. */ 969 robj *o = lookupKeyWrite(rl->db,rl->key); 970 if (o != NULL && o->type == REDIS_LIST) { 971 dictEntry *de; 972 973 /* We serve clients in the same order they blocked for 974 * this key, from the first blocked to the last. */ 975 de = dictFind(rl->db->blocking_keys,rl->key); 976 if (de) { 977 list *clients = dictGetVal(de); 978 int numclients = listLength(clients); 979 980 while(numclients--) { 981 listNode *clientnode = listFirst(clients); 982 redisClient *receiver = clientnode->value; 983 robj *dstkey = receiver->bpop.target; 984 int where = (receiver->lastcmd && 985 receiver->lastcmd->proc == blpopCommand) ? 986 REDIS_HEAD : REDIS_TAIL; 987 robj *value = listTypePop(o,where); 988 989 if (value) { 990 /* Protect receiver->bpop.target, that will be 991 * freed by the next unblockClient() 992 * call. */ 993 if (dstkey) incrRefCount(dstkey); 994 unblockClient(receiver); 995 996 if (serveClientBlockedOnList(receiver, 997 rl->key,dstkey,rl->db,value, 998 where) == REDIS_ERR) 999 { 1000 /* If we failed serving the client we need 1001 * to also undo the POP operation. */ 1002 listTypePush(o,value,where); 1003 } 1004 1005 if (dstkey) decrRefCount(dstkey); 1006 decrRefCount(value); 1007 } else { 1008 break; 1009 } 1010 } 1011 } 1012 1013 if (listTypeLength(o) == 0) dbDelete(rl->db,rl->key); 1014 /* We don't call signalModifiedKey() as it was already called 1015 * when an element was pushed on the list. */ 1016 } 1017 1018 /* Free this item. */ 1019 decrRefCount(rl->key); 1020 zfree(rl); 1021 listDelNode(l,ln); 1022 } 1023 listRelease(l); /* We have the new list on place at this point. */ 1024 } 1025 } 1026 1027 /* Blocking RPOP/LPOP */ 1028 void blockingPopGenericCommand(redisClient *c, int where) { 1029 robj *o; 1030 mstime_t timeout; 1031 int j; 1032 1033 if (getTimeoutFromObjectOrReply(c,c->argv[c->argc-1],&timeout,UNIT_SECONDS) 1034 != REDIS_OK) return; 1035 1036 for (j = 1; j < c->argc-1; j++) { 1037 o = lookupKeyWrite(c->db,c->argv[j]); 1038 if (o != NULL) { 1039 if (o->type != REDIS_LIST) { 1040 addReply(c,shared.wrongtypeerr); 1041 return; 1042 } else { 1043 if (listTypeLength(o) != 0) { 1044 /* Non empty list, this is like a non normal [LR]POP. */ 1045 char *event = (where == REDIS_HEAD) ? "lpop" : "rpop"; 1046 robj *value = listTypePop(o,where); 1047 redisAssert(value != NULL); 1048 1049 addReplyMultiBulkLen(c,2); 1050 addReplyBulk(c,c->argv[j]); 1051 addReplyBulk(c,value); 1052 decrRefCount(value); 1053 notifyKeyspaceEvent(REDIS_NOTIFY_LIST,event, 1054 c->argv[j],c->db->id); 1055 if (listTypeLength(o) == 0) { 1056 dbDelete(c->db,c->argv[j]); 1057 notifyKeyspaceEvent(REDIS_NOTIFY_GENERIC,"del", 1058 c->argv[j],c->db->id); 1059 } 1060 signalModifiedKey(c->db,c->argv[j]); 1061 server.dirty++; 1062 1063 /* Replicate it as an [LR]POP instead of B[LR]POP. */ 1064 rewriteClientCommandVector(c,2, 1065 (where == REDIS_HEAD) ? shared.lpop : shared.rpop, 1066 c->argv[j]); 1067 return; 1068 } 1069 } 1070 } 1071 } 1072 1073 /* If we are inside a MULTI/EXEC and the list is empty the only thing 1074 * we can do is treating it as a timeout (even with timeout 0). */ 1075 if (c->flags & REDIS_MULTI) { 1076 addReply(c,shared.nullmultibulk); 1077 return; 1078 } 1079 1080 /* If the list is empty or the key does not exists we must block */ 1081 blockForKeys(c, c->argv + 1, c->argc - 2, timeout, NULL); 1082 } 1083 1084 void blpopCommand(redisClient *c) { 1085 blockingPopGenericCommand(c,REDIS_HEAD); 1086 } 1087 1088 void brpopCommand(redisClient *c) { 1089 blockingPopGenericCommand(c,REDIS_TAIL); 1090 } 1091 1092 void brpoplpushCommand(redisClient *c) { 1093 mstime_t timeout; 1094 1095 if (getTimeoutFromObjectOrReply(c,c->argv[3],&timeout,UNIT_SECONDS) 1096 != REDIS_OK) return; 1097 1098 robj *key = lookupKeyWrite(c->db, c->argv[1]); 1099 1100 if (key == NULL) { 1101 if (c->flags & REDIS_MULTI) { 1102 /* Blocking against an empty list in a multi state 1103 * returns immediately. */ 1104 addReply(c, shared.nullbulk); 1105 } else { 1106 /* The list is empty and the client blocks. */ 1107 blockForKeys(c, c->argv + 1, 1, timeout, c->argv[2]); 1108 } 1109 } else { 1110 if (key->type != REDIS_LIST) { 1111 addReply(c, shared.wrongtypeerr); 1112 } else { 1113 /* The list exists and has elements, so 1114 * the regular rpoplpushCommand is executed. */ 1115 redisAssertWithInfo(c,key,listTypeLength(key) > 0); 1116 rpoplpushCommand(c); 1117 } 1118 } 1119 } 1120