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