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 (sdsEncodedObject(value) && 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,sdsEncodedObject(o)); 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(REDIS_NOTIFY_LIST,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(REDIS_NOTIFY_LIST,"linsert", 371 c->argv[1],c->db->id); 372 server.dirty++; 373 } else { 374 /* Notify client of a failed insert */ 375 addReply(c,shared.cnegone); 376 return; 377 } 378 } else { 379 char *event = (where == REDIS_HEAD) ? "lpush" : "rpush"; 380 381 listTypePush(subject,val,where); 382 signalModifiedKey(c->db,c->argv[1]); 383 notifyKeyspaceEvent(REDIS_NOTIFY_LIST,event,c->argv[1],c->db->id); 384 server.dirty++; 385 } 386 387 addReplyLongLong(c,listTypeLength(subject)); 388 } 389 390 void lpushxCommand(redisClient *c) { 391 c->argv[2] = tryObjectEncoding(c->argv[2]); 392 pushxGenericCommand(c,NULL,c->argv[2],REDIS_HEAD); 393 } 394 395 void rpushxCommand(redisClient *c) { 396 c->argv[2] = tryObjectEncoding(c->argv[2]); 397 pushxGenericCommand(c,NULL,c->argv[2],REDIS_TAIL); 398 } 399 400 void linsertCommand(redisClient *c) { 401 c->argv[4] = tryObjectEncoding(c->argv[4]); 402 if (strcasecmp(c->argv[2]->ptr,"after") == 0) { 403 pushxGenericCommand(c,c->argv[3],c->argv[4],REDIS_TAIL); 404 } else if (strcasecmp(c->argv[2]->ptr,"before") == 0) { 405 pushxGenericCommand(c,c->argv[3],c->argv[4],REDIS_HEAD); 406 } else { 407 addReply(c,shared.syntaxerr); 408 } 409 } 410 411 void llenCommand(redisClient *c) { 412 robj *o = lookupKeyReadOrReply(c,c->argv[1],shared.czero); 413 if (o == NULL || checkType(c,o,REDIS_LIST)) return; 414 addReplyLongLong(c,listTypeLength(o)); 415 } 416 417 void lindexCommand(redisClient *c) { 418 robj *o = lookupKeyReadOrReply(c,c->argv[1],shared.nullbulk); 419 if (o == NULL || checkType(c,o,REDIS_LIST)) return; 420 long index; 421 robj *value = NULL; 422 423 if ((getLongFromObjectOrReply(c, c->argv[2], &index, NULL) != REDIS_OK)) 424 return; 425 426 if (o->encoding == REDIS_ENCODING_ZIPLIST) { 427 unsigned char *p; 428 unsigned char *vstr; 429 unsigned int vlen; 430 long long vlong; 431 p = ziplistIndex(o->ptr,index); 432 if (ziplistGet(p,&vstr,&vlen,&vlong)) { 433 if (vstr) { 434 value = createStringObject((char*)vstr,vlen); 435 } else { 436 value = createStringObjectFromLongLong(vlong); 437 } 438 addReplyBulk(c,value); 439 decrRefCount(value); 440 } else { 441 addReply(c,shared.nullbulk); 442 } 443 } else if (o->encoding == REDIS_ENCODING_LINKEDLIST) { 444 listNode *ln = listIndex(o->ptr,index); 445 if (ln != NULL) { 446 value = listNodeValue(ln); 447 addReplyBulk(c,value); 448 } else { 449 addReply(c,shared.nullbulk); 450 } 451 } else { 452 redisPanic("Unknown list encoding"); 453 } 454 } 455 456 void lsetCommand(redisClient *c) { 457 robj *o = lookupKeyWriteOrReply(c,c->argv[1],shared.nokeyerr); 458 if (o == NULL || checkType(c,o,REDIS_LIST)) return; 459 long index; 460 robj *value = (c->argv[3] = tryObjectEncoding(c->argv[3])); 461 462 if ((getLongFromObjectOrReply(c, c->argv[2], &index, NULL) != REDIS_OK)) 463 return; 464 465 listTypeTryConversion(o,value); 466 if (o->encoding == REDIS_ENCODING_ZIPLIST) { 467 unsigned char *p, *zl = o->ptr; 468 p = ziplistIndex(zl,index); 469 if (p == NULL) { 470 addReply(c,shared.outofrangeerr); 471 } else { 472 o->ptr = ziplistDelete(o->ptr,&p); 473 value = getDecodedObject(value); 474 o->ptr = ziplistInsert(o->ptr,p,value->ptr,sdslen(value->ptr)); 475 decrRefCount(value); 476 addReply(c,shared.ok); 477 signalModifiedKey(c->db,c->argv[1]); 478 notifyKeyspaceEvent(REDIS_NOTIFY_LIST,"lset",c->argv[1],c->db->id); 479 server.dirty++; 480 } 481 } else if (o->encoding == REDIS_ENCODING_LINKEDLIST) { 482 listNode *ln = listIndex(o->ptr,index); 483 if (ln == NULL) { 484 addReply(c,shared.outofrangeerr); 485 } else { 486 decrRefCount((robj*)listNodeValue(ln)); 487 listNodeValue(ln) = value; 488 incrRefCount(value); 489 addReply(c,shared.ok); 490 signalModifiedKey(c->db,c->argv[1]); 491 notifyKeyspaceEvent(REDIS_NOTIFY_LIST,"lset",c->argv[1],c->db->id); 492 server.dirty++; 493 } 494 } else { 495 redisPanic("Unknown list encoding"); 496 } 497 } 498 499 void popGenericCommand(redisClient *c, int where) { 500 robj *o = lookupKeyWriteOrReply(c,c->argv[1],shared.nullbulk); 501 if (o == NULL || checkType(c,o,REDIS_LIST)) return; 502 503 robj *value = listTypePop(o,where); 504 if (value == NULL) { 505 addReply(c,shared.nullbulk); 506 } else { 507 char *event = (where == REDIS_HEAD) ? "lpop" : "rpop"; 508 509 addReplyBulk(c,value); 510 decrRefCount(value); 511 notifyKeyspaceEvent(REDIS_NOTIFY_LIST,event,c->argv[1],c->db->id); 512 if (listTypeLength(o) == 0) { 513 notifyKeyspaceEvent(REDIS_NOTIFY_GENERIC,"del", 514 c->argv[1],c->db->id); 515 dbDelete(c->db,c->argv[1]); 516 } 517 signalModifiedKey(c->db,c->argv[1]); 518 server.dirty++; 519 } 520 } 521 522 void lpopCommand(redisClient *c) { 523 popGenericCommand(c,REDIS_HEAD); 524 } 525 526 void rpopCommand(redisClient *c) { 527 popGenericCommand(c,REDIS_TAIL); 528 } 529 530 void lrangeCommand(redisClient *c) { 531 robj *o; 532 long start, end, llen, rangelen; 533 534 if ((getLongFromObjectOrReply(c, c->argv[2], &start, NULL) != REDIS_OK) || 535 (getLongFromObjectOrReply(c, c->argv[3], &end, NULL) != REDIS_OK)) return; 536 537 if ((o = lookupKeyReadOrReply(c,c->argv[1],shared.emptymultibulk)) == NULL 538 || checkType(c,o,REDIS_LIST)) return; 539 llen = listTypeLength(o); 540 541 /* convert negative indexes */ 542 if (start < 0) start = llen+start; 543 if (end < 0) end = llen+end; 544 if (start < 0) start = 0; 545 546 /* Invariant: start >= 0, so this test will be true when end < 0. 547 * The range is empty when start > end or start >= length. */ 548 if (start > end || start >= llen) { 549 addReply(c,shared.emptymultibulk); 550 return; 551 } 552 if (end >= llen) end = llen-1; 553 rangelen = (end-start)+1; 554 555 /* Return the result in form of a multi-bulk reply */ 556 addReplyMultiBulkLen(c,rangelen); 557 if (o->encoding == REDIS_ENCODING_ZIPLIST) { 558 unsigned char *p = ziplistIndex(o->ptr,start); 559 unsigned char *vstr; 560 unsigned int vlen; 561 long long vlong; 562 563 while(rangelen--) { 564 ziplistGet(p,&vstr,&vlen,&vlong); 565 if (vstr) { 566 addReplyBulkCBuffer(c,vstr,vlen); 567 } else { 568 addReplyBulkLongLong(c,vlong); 569 } 570 p = ziplistNext(o->ptr,p); 571 } 572 } else if (o->encoding == REDIS_ENCODING_LINKEDLIST) { 573 listNode *ln; 574 575 /* If we are nearest to the end of the list, reach the element 576 * starting from tail and going backward, as it is faster. */ 577 if (start > llen/2) start -= llen; 578 ln = listIndex(o->ptr,start); 579 580 while(rangelen--) { 581 addReplyBulk(c,ln->value); 582 ln = ln->next; 583 } 584 } else { 585 redisPanic("List encoding is not LINKEDLIST nor ZIPLIST!"); 586 } 587 } 588 589 void ltrimCommand(redisClient *c) { 590 robj *o; 591 long start, end, llen, j, ltrim, rtrim; 592 list *list; 593 listNode *ln; 594 595 if ((getLongFromObjectOrReply(c, c->argv[2], &start, NULL) != REDIS_OK) || 596 (getLongFromObjectOrReply(c, c->argv[3], &end, NULL) != REDIS_OK)) return; 597 598 if ((o = lookupKeyWriteOrReply(c,c->argv[1],shared.ok)) == NULL || 599 checkType(c,o,REDIS_LIST)) return; 600 llen = listTypeLength(o); 601 602 /* convert negative indexes */ 603 if (start < 0) start = llen+start; 604 if (end < 0) end = llen+end; 605 if (start < 0) start = 0; 606 607 /* Invariant: start >= 0, so this test will be true when end < 0. 608 * The range is empty when start > end or start >= length. */ 609 if (start > end || start >= llen) { 610 /* Out of range start or start > end result in empty list */ 611 ltrim = llen; 612 rtrim = 0; 613 } else { 614 if (end >= llen) end = llen-1; 615 ltrim = start; 616 rtrim = llen-end-1; 617 } 618 619 /* Remove list elements to perform the trim */ 620 if (o->encoding == REDIS_ENCODING_ZIPLIST) { 621 o->ptr = ziplistDeleteRange(o->ptr,0,ltrim); 622 o->ptr = ziplistDeleteRange(o->ptr,-rtrim,rtrim); 623 } else if (o->encoding == REDIS_ENCODING_LINKEDLIST) { 624 list = o->ptr; 625 for (j = 0; j < ltrim; j++) { 626 ln = listFirst(list); 627 listDelNode(list,ln); 628 } 629 for (j = 0; j < rtrim; j++) { 630 ln = listLast(list); 631 listDelNode(list,ln); 632 } 633 } else { 634 redisPanic("Unknown list encoding"); 635 } 636 637 notifyKeyspaceEvent(REDIS_NOTIFY_LIST,"ltrim",c->argv[1],c->db->id); 638 if (listTypeLength(o) == 0) { 639 dbDelete(c->db,c->argv[1]); 640 notifyKeyspaceEvent(REDIS_NOTIFY_GENERIC,"del",c->argv[1],c->db->id); 641 } 642 signalModifiedKey(c->db,c->argv[1]); 643 server.dirty++; 644 addReply(c,shared.ok); 645 } 646 647 void lremCommand(redisClient *c) { 648 robj *subject, *obj; 649 obj = c->argv[3] = tryObjectEncoding(c->argv[3]); 650 long toremove; 651 long removed = 0; 652 listTypeEntry entry; 653 654 if ((getLongFromObjectOrReply(c, c->argv[2], &toremove, NULL) != REDIS_OK)) 655 return; 656 657 subject = lookupKeyWriteOrReply(c,c->argv[1],shared.czero); 658 if (subject == NULL || checkType(c,subject,REDIS_LIST)) return; 659 660 /* Make sure obj is raw when we're dealing with a ziplist */ 661 if (subject->encoding == REDIS_ENCODING_ZIPLIST) 662 obj = getDecodedObject(obj); 663 664 listTypeIterator *li; 665 if (toremove < 0) { 666 toremove = -toremove; 667 li = listTypeInitIterator(subject,-1,REDIS_HEAD); 668 } else { 669 li = listTypeInitIterator(subject,0,REDIS_TAIL); 670 } 671 672 while (listTypeNext(li,&entry)) { 673 if (listTypeEqual(&entry,obj)) { 674 listTypeDelete(&entry); 675 server.dirty++; 676 removed++; 677 if (toremove && removed == toremove) break; 678 } 679 } 680 listTypeReleaseIterator(li); 681 682 /* Clean up raw encoded object */ 683 if (subject->encoding == REDIS_ENCODING_ZIPLIST) 684 decrRefCount(obj); 685 686 if (listTypeLength(subject) == 0) dbDelete(c->db,c->argv[1]); 687 addReplyLongLong(c,removed); 688 if (removed) signalModifiedKey(c->db,c->argv[1]); 689 } 690 691 /* This is the semantic of this command: 692 * RPOPLPUSH srclist dstlist: 693 * IF LLEN(srclist) > 0 694 * element = RPOP srclist 695 * LPUSH dstlist element 696 * RETURN element 697 * ELSE 698 * RETURN nil 699 * END 700 * END 701 * 702 * The idea is to be able to get an element from a list in a reliable way 703 * since the element is not just returned but pushed against another list 704 * as well. This command was originally proposed by Ezra Zygmuntowicz. 705 */ 706 707 void rpoplpushHandlePush(redisClient *c, robj *dstkey, robj *dstobj, robj *value) { 708 /* Create the list if the key does not exist */ 709 if (!dstobj) { 710 dstobj = createZiplistObject(); 711 dbAdd(c->db,dstkey,dstobj); 712 signalListAsReady(c,dstkey); 713 } 714 signalModifiedKey(c->db,dstkey); 715 listTypePush(dstobj,value,REDIS_HEAD); 716 notifyKeyspaceEvent(REDIS_NOTIFY_LIST,"lpush",dstkey,c->db->id); 717 /* Always send the pushed value to the client. */ 718 addReplyBulk(c,value); 719 } 720 721 void rpoplpushCommand(redisClient *c) { 722 robj *sobj, *value; 723 if ((sobj = lookupKeyWriteOrReply(c,c->argv[1],shared.nullbulk)) == NULL || 724 checkType(c,sobj,REDIS_LIST)) return; 725 726 if (listTypeLength(sobj) == 0) { 727 /* This may only happen after loading very old RDB files. Recent 728 * versions of Redis delete keys of empty lists. */ 729 addReply(c,shared.nullbulk); 730 } else { 731 robj *dobj = lookupKeyWrite(c->db,c->argv[2]); 732 robj *touchedkey = c->argv[1]; 733 734 if (dobj && checkType(c,dobj,REDIS_LIST)) return; 735 value = listTypePop(sobj,REDIS_TAIL); 736 /* We saved touched key, and protect it, since rpoplpushHandlePush 737 * may change the client command argument vector (it does not 738 * currently). */ 739 incrRefCount(touchedkey); 740 rpoplpushHandlePush(c,c->argv[2],dobj,value); 741 742 /* listTypePop returns an object with its refcount incremented */ 743 decrRefCount(value); 744 745 /* Delete the source list when it is empty */ 746 notifyKeyspaceEvent(REDIS_NOTIFY_LIST,"rpop",touchedkey,c->db->id); 747 if (listTypeLength(sobj) == 0) { 748 dbDelete(c->db,touchedkey); 749 notifyKeyspaceEvent(REDIS_NOTIFY_GENERIC,"del", 750 touchedkey,c->db->id); 751 } 752 signalModifiedKey(c->db,touchedkey); 753 decrRefCount(touchedkey); 754 server.dirty++; 755 } 756 } 757 758 /*----------------------------------------------------------------------------- 759 * Blocking POP operations 760 *----------------------------------------------------------------------------*/ 761 762 /* This is how the current blocking POP works, we use BLPOP as example: 763 * - If the user calls BLPOP and the key exists and contains a non empty list 764 * then LPOP is called instead. So BLPOP is semantically the same as LPOP 765 * if blocking is not required. 766 * - If instead BLPOP is called and the key does not exists or the list is 767 * empty we need to block. In order to do so we remove the notification for 768 * new data to read in the client socket (so that we'll not serve new 769 * requests if the blocking request is not served). Also we put the client 770 * in a dictionary (db->blocking_keys) mapping keys to a list of clients 771 * blocking for this keys. 772 * - If a PUSH operation against a key with blocked clients waiting is 773 * performed, we mark this key as "ready", and after the current command, 774 * MULTI/EXEC block, or script, is executed, we serve all the clients waiting 775 * for this list, from the one that blocked first, to the last, accordingly 776 * to the number of elements we have in the ready list. 777 */ 778 779 /* Set a client in blocking mode for the specified key, with the specified 780 * timeout */ 781 void blockForKeys(redisClient *c, robj **keys, int numkeys, mstime_t timeout, robj *target) { 782 dictEntry *de; 783 list *l; 784 int j; 785 786 c->bpop.timeout = timeout; 787 c->bpop.target = target; 788 789 if (target != NULL) incrRefCount(target); 790 791 for (j = 0; j < numkeys; j++) { 792 /* If the key already exists in the dict ignore it. */ 793 if (dictAdd(c->bpop.keys,keys[j],NULL) != DICT_OK) continue; 794 incrRefCount(keys[j]); 795 796 /* And in the other "side", to map keys -> clients */ 797 de = dictFind(c->db->blocking_keys,keys[j]); 798 if (de == NULL) { 799 int retval; 800 801 /* For every key we take a list of clients blocked for it */ 802 l = listCreate(); 803 retval = dictAdd(c->db->blocking_keys,keys[j],l); 804 incrRefCount(keys[j]); 805 redisAssertWithInfo(c,keys[j],retval == DICT_OK); 806 } else { 807 l = dictGetVal(de); 808 } 809 listAddNodeTail(l,c); 810 } 811 blockClient(c,REDIS_BLOCKED_LIST); 812 } 813 814 /* Unblock a client that's waiting in a blocking operation such as BLPOP. 815 * You should never call this function directly, but unblockClient() instead. */ 816 void unblockClientWaitingData(redisClient *c) { 817 dictEntry *de; 818 dictIterator *di; 819 list *l; 820 821 redisAssertWithInfo(c,NULL,dictSize(c->bpop.keys) != 0); 822 di = dictGetIterator(c->bpop.keys); 823 /* The client may wait for multiple keys, so unblock it for every key. */ 824 while((de = dictNext(di)) != NULL) { 825 robj *key = dictGetKey(de); 826 827 /* Remove this client from the list of clients waiting for this key. */ 828 l = dictFetchValue(c->db->blocking_keys,key); 829 redisAssertWithInfo(c,key,l != NULL); 830 listDelNode(l,listSearchKey(l,c)); 831 /* If the list is empty we need to remove it to avoid wasting memory */ 832 if (listLength(l) == 0) 833 dictDelete(c->db->blocking_keys,key); 834 } 835 dictReleaseIterator(di); 836 837 /* Cleanup the client structure */ 838 dictEmpty(c->bpop.keys,NULL); 839 if (c->bpop.target) { 840 decrRefCount(c->bpop.target); 841 c->bpop.target = NULL; 842 } 843 } 844 845 /* If the specified key has clients blocked waiting for list pushes, this 846 * function will put the key reference into the server.ready_keys list. 847 * Note that db->ready_keys is a hash table that allows us to avoid putting 848 * the same key again and again in the list in case of multiple pushes 849 * made by a script or in the context of MULTI/EXEC. 850 * 851 * The list will be finally processed by handleClientsBlockedOnLists() */ 852 void signalListAsReady(redisClient *c, robj *key) { 853 readyList *rl; 854 855 /* No clients blocking for this key? No need to queue it. */ 856 if (dictFind(c->db->blocking_keys,key) == NULL) return; 857 858 /* Key was already signaled? No need to queue it again. */ 859 if (dictFind(c->db->ready_keys,key) != NULL) return; 860 861 /* Ok, we need to queue this key into server.ready_keys. */ 862 rl = zmalloc(sizeof(*rl)); 863 rl->key = key; 864 rl->db = c->db; 865 incrRefCount(key); 866 listAddNodeTail(server.ready_keys,rl); 867 868 /* We also add the key in the db->ready_keys dictionary in order 869 * to avoid adding it multiple times into a list with a simple O(1) 870 * check. */ 871 incrRefCount(key); 872 redisAssert(dictAdd(c->db->ready_keys,key,NULL) == DICT_OK); 873 } 874 875 /* This is a helper function for handleClientsBlockedOnLists(). It's work 876 * is to serve a specific client (receiver) that is blocked on 'key' 877 * in the context of the specified 'db', doing the following: 878 * 879 * 1) Provide the client with the 'value' element. 880 * 2) If the dstkey is not NULL (we are serving a BRPOPLPUSH) also push the 881 * 'value' element on the destination list (the LPUSH side of the command). 882 * 3) Propagate the resulting BRPOP, BLPOP and additional LPUSH if any into 883 * the AOF and replication channel. 884 * 885 * The argument 'where' is REDIS_TAIL or REDIS_HEAD, and indicates if the 886 * 'value' element was popped fron the head (BLPOP) or tail (BRPOP) so that 887 * we can propagate the command properly. 888 * 889 * The function returns REDIS_OK if we are able to serve the client, otherwise 890 * REDIS_ERR is returned to signal the caller that the list POP operation 891 * should be undone as the client was not served: This only happens for 892 * BRPOPLPUSH that fails to push the value to the destination key as it is 893 * of the wrong type. */ 894 int serveClientBlockedOnList(redisClient *receiver, robj *key, robj *dstkey, redisDb *db, robj *value, int where) 895 { 896 robj *argv[3]; 897 898 if (dstkey == NULL) { 899 /* Propagate the [LR]POP operation. */ 900 argv[0] = (where == REDIS_HEAD) ? shared.lpop : 901 shared.rpop; 902 argv[1] = key; 903 propagate((where == REDIS_HEAD) ? 904 server.lpopCommand : server.rpopCommand, 905 db->id,argv,2,REDIS_PROPAGATE_AOF|REDIS_PROPAGATE_REPL); 906 907 /* BRPOP/BLPOP */ 908 addReplyMultiBulkLen(receiver,2); 909 addReplyBulk(receiver,key); 910 addReplyBulk(receiver,value); 911 } else { 912 /* BRPOPLPUSH */ 913 robj *dstobj = 914 lookupKeyWrite(receiver->db,dstkey); 915 if (!(dstobj && 916 checkType(receiver,dstobj,REDIS_LIST))) 917 { 918 /* Propagate the RPOP operation. */ 919 argv[0] = shared.rpop; 920 argv[1] = key; 921 propagate(server.rpopCommand, 922 db->id,argv,2, 923 REDIS_PROPAGATE_AOF| 924 REDIS_PROPAGATE_REPL); 925 rpoplpushHandlePush(receiver,dstkey,dstobj, 926 value); 927 /* Propagate the LPUSH operation. */ 928 argv[0] = shared.lpush; 929 argv[1] = dstkey; 930 argv[2] = value; 931 propagate(server.lpushCommand, 932 db->id,argv,3, 933 REDIS_PROPAGATE_AOF| 934 REDIS_PROPAGATE_REPL); 935 } else { 936 /* BRPOPLPUSH failed because of wrong 937 * destination type. */ 938 return REDIS_ERR; 939 } 940 } 941 return REDIS_OK; 942 } 943 944 /* This function should be called by Redis every time a single command, 945 * a MULTI/EXEC block, or a Lua script, terminated its execution after 946 * being called by a client. 947 * 948 * All the keys with at least one client blocked that received at least 949 * one new element via some PUSH operation are accumulated into 950 * the server.ready_keys list. This function will run the list and will 951 * serve clients accordingly. Note that the function will iterate again and 952 * again as a result of serving BRPOPLPUSH we can have new blocking clients 953 * to serve because of the PUSH side of BRPOPLPUSH. */ 954 void handleClientsBlockedOnLists(void) { 955 while(listLength(server.ready_keys) != 0) { 956 list *l; 957 958 /* Point server.ready_keys to a fresh list and save the current one 959 * locally. This way as we run the old list we are free to call 960 * signalListAsReady() that may push new elements in server.ready_keys 961 * when handling clients blocked into BRPOPLPUSH. */ 962 l = server.ready_keys; 963 server.ready_keys = listCreate(); 964 965 while(listLength(l) != 0) { 966 listNode *ln = listFirst(l); 967 readyList *rl = ln->value; 968 969 /* First of all remove this key from db->ready_keys so that 970 * we can safely call signalListAsReady() against this key. */ 971 dictDelete(rl->db->ready_keys,rl->key); 972 973 /* If the key exists and it's a list, serve blocked clients 974 * with data. */ 975 robj *o = lookupKeyWrite(rl->db,rl->key); 976 if (o != NULL && o->type == REDIS_LIST) { 977 dictEntry *de; 978 979 /* We serve clients in the same order they blocked for 980 * this key, from the first blocked to the last. */ 981 de = dictFind(rl->db->blocking_keys,rl->key); 982 if (de) { 983 list *clients = dictGetVal(de); 984 int numclients = listLength(clients); 985 986 while(numclients--) { 987 listNode *clientnode = listFirst(clients); 988 redisClient *receiver = clientnode->value; 989 robj *dstkey = receiver->bpop.target; 990 int where = (receiver->lastcmd && 991 receiver->lastcmd->proc == blpopCommand) ? 992 REDIS_HEAD : REDIS_TAIL; 993 robj *value = listTypePop(o,where); 994 995 if (value) { 996 /* Protect receiver->bpop.target, that will be 997 * freed by the next unblockClient() 998 * call. */ 999 if (dstkey) incrRefCount(dstkey); 1000 unblockClient(receiver); 1001 1002 if (serveClientBlockedOnList(receiver, 1003 rl->key,dstkey,rl->db,value, 1004 where) == REDIS_ERR) 1005 { 1006 /* If we failed serving the client we need 1007 * to also undo the POP operation. */ 1008 listTypePush(o,value,where); 1009 } 1010 1011 if (dstkey) decrRefCount(dstkey); 1012 decrRefCount(value); 1013 } else { 1014 break; 1015 } 1016 } 1017 } 1018 1019 if (listTypeLength(o) == 0) dbDelete(rl->db,rl->key); 1020 /* We don't call signalModifiedKey() as it was already called 1021 * when an element was pushed on the list. */ 1022 } 1023 1024 /* Free this item. */ 1025 decrRefCount(rl->key); 1026 zfree(rl); 1027 listDelNode(l,ln); 1028 } 1029 listRelease(l); /* We have the new list on place at this point. */ 1030 } 1031 } 1032 1033 /* Blocking RPOP/LPOP */ 1034 void blockingPopGenericCommand(redisClient *c, int where) { 1035 robj *o; 1036 mstime_t timeout; 1037 int j; 1038 1039 if (getTimeoutFromObjectOrReply(c,c->argv[c->argc-1],&timeout,UNIT_SECONDS) 1040 != REDIS_OK) return; 1041 1042 for (j = 1; j < c->argc-1; j++) { 1043 o = lookupKeyWrite(c->db,c->argv[j]); 1044 if (o != NULL) { 1045 if (o->type != REDIS_LIST) { 1046 addReply(c,shared.wrongtypeerr); 1047 return; 1048 } else { 1049 if (listTypeLength(o) != 0) { 1050 /* Non empty list, this is like a non normal [LR]POP. */ 1051 char *event = (where == REDIS_HEAD) ? "lpop" : "rpop"; 1052 robj *value = listTypePop(o,where); 1053 redisAssert(value != NULL); 1054 1055 addReplyMultiBulkLen(c,2); 1056 addReplyBulk(c,c->argv[j]); 1057 addReplyBulk(c,value); 1058 decrRefCount(value); 1059 notifyKeyspaceEvent(REDIS_NOTIFY_LIST,event, 1060 c->argv[j],c->db->id); 1061 if (listTypeLength(o) == 0) { 1062 dbDelete(c->db,c->argv[j]); 1063 notifyKeyspaceEvent(REDIS_NOTIFY_GENERIC,"del", 1064 c->argv[j],c->db->id); 1065 } 1066 signalModifiedKey(c->db,c->argv[j]); 1067 server.dirty++; 1068 1069 /* Replicate it as an [LR]POP instead of B[LR]POP. */ 1070 rewriteClientCommandVector(c,2, 1071 (where == REDIS_HEAD) ? shared.lpop : shared.rpop, 1072 c->argv[j]); 1073 return; 1074 } 1075 } 1076 } 1077 } 1078 1079 /* If we are inside a MULTI/EXEC and the list is empty the only thing 1080 * we can do is treating it as a timeout (even with timeout 0). */ 1081 if (c->flags & REDIS_MULTI) { 1082 addReply(c,shared.nullmultibulk); 1083 return; 1084 } 1085 1086 /* If the list is empty or the key does not exists we must block */ 1087 blockForKeys(c, c->argv + 1, c->argc - 2, timeout, NULL); 1088 } 1089 1090 void blpopCommand(redisClient *c) { 1091 blockingPopGenericCommand(c,REDIS_HEAD); 1092 } 1093 1094 void brpopCommand(redisClient *c) { 1095 blockingPopGenericCommand(c,REDIS_TAIL); 1096 } 1097 1098 void brpoplpushCommand(redisClient *c) { 1099 mstime_t timeout; 1100 1101 if (getTimeoutFromObjectOrReply(c,c->argv[3],&timeout,UNIT_SECONDS) 1102 != REDIS_OK) return; 1103 1104 robj *key = lookupKeyWrite(c->db, c->argv[1]); 1105 1106 if (key == NULL) { 1107 if (c->flags & REDIS_MULTI) { 1108 /* Blocking against an empty list in a multi state 1109 * returns immediately. */ 1110 addReply(c, shared.nullbulk); 1111 } else { 1112 /* The list is empty and the client blocks. */ 1113 blockForKeys(c, c->argv + 1, 1, timeout, c->argv[2]); 1114 } 1115 } else { 1116 if (key->type != REDIS_LIST) { 1117 addReply(c, shared.wrongtypeerr); 1118 } else { 1119 /* The list exists and has elements, so 1120 * the regular rpoplpushCommand is executed. */ 1121 redisAssertWithInfo(c,key,listTypeLength(key) > 0); 1122 rpoplpushCommand(c); 1123 } 1124 } 1125 } 1126