src/t_list.c

   1 #include "redis.h"
   2
   3 /*-----------------------------------------------------------------------------
   4  * List API
   5  *----------------------------------------------------------------------------*/
   6
   7 /* Check the argument length to see if it requires us to convert the ziplist
   8  * to a real list. Only check raw-encoded objects because integer encoded
   9  * objects are never too long. */
  10 void listTypeTryConversion(robj *subject, robj *value) {
  11     if (subject->encoding != REDIS_ENCODING_ZIPLIST) return;
  12     if (value->encoding == REDIS_ENCODING_RAW &&
  13         sdslen(value->ptr) > server.list_max_ziplist_value)
  14             listTypeConvert(subject,REDIS_ENCODING_LINKEDLIST);
  15 }
  16
  17 void listTypePush(robj *subject, robj *value, int where) {
  18     /* Check if we need to convert the ziplist */
  19     listTypeTryConversion(subject,value);
  20     if (subject->encoding == REDIS_ENCODING_ZIPLIST &&
  21         ziplistLen(subject->ptr) >= server.list_max_ziplist_entries)
  22             listTypeConvert(subject,REDIS_ENCODING_LINKEDLIST);
  23
  24     if (subject->encoding == REDIS_ENCODING_ZIPLIST) {
  25         int pos = (where == REDIS_HEAD) ? ZIPLIST_HEAD : ZIPLIST_TAIL;
  26         value = getDecodedObject(value);
  27         subject->ptr = ziplistPush(subject->ptr,value->ptr,sdslen(value->ptr),pos);
  28         decrRefCount(value);
  29     } else if (subject->encoding == REDIS_ENCODING_LINKEDLIST) {
  30         if (where == REDIS_HEAD) {
  31             listAddNodeHead(subject->ptr,value);
  32         } else {
  33             listAddNodeTail(subject->ptr,value);
  34         }
  35         incrRefCount(value);
  36     } else {
  37         redisPanic("Unknown list encoding");
  38     }
  39 }
  40
  41 robj *listTypePop(robj *subject, int where) {
  42     robj *value = NULL;
  43     if (subject->encoding == REDIS_ENCODING_ZIPLIST) {
  44         unsigned char *p;
  45         unsigned char *vstr;
  46         unsigned int vlen;
  47         long long vlong;
  48         int pos = (where == REDIS_HEAD) ? 0 : -1;
  49         p = ziplistIndex(subject->ptr,pos);
  50         if (ziplistGet(p,&vstr,&vlen,&vlong)) {
  51             if (vstr) {
  52                 value = createStringObject((char*)vstr,vlen);
  53             } else {
  54                 value = createStringObjectFromLongLong(vlong);
  55             }
  56             /* We only need to delete an element when it exists */
  57             subject->ptr = ziplistDelete(subject->ptr,&p);
  58         }
  59     } else if (subject->encoding == REDIS_ENCODING_LINKEDLIST) {
  60         list *list = subject->ptr;
  61         listNode *ln;
  62         if (where == REDIS_HEAD) {
  63             ln = listFirst(list);
  64         } else {
  65             ln = listLast(list);
  66         }
  67         if (ln != NULL) {
  68             value = listNodeValue(ln);
  69             incrRefCount(value);
  70             listDelNode(list,ln);
  71         }
  72     } else {
  73         redisPanic("Unknown list encoding");
  74     }
  75     return value;
  76 }
  77
  78 unsigned long listTypeLength(robj *subject) {
  79     if (subject->encoding == REDIS_ENCODING_ZIPLIST) {
  80         return ziplistLen(subject->ptr);
  81     } else if (subject->encoding == REDIS_ENCODING_LINKEDLIST) {
  82         return listLength((list*)subject->ptr);
  83     } else {
  84         redisPanic("Unknown list encoding");
  85     }
  86 }
  87
  88 /* Initialize an iterator at the specified index. */
  89 listTypeIterator *listTypeInitIterator(robj *subject, int index, unsigned char direction) {
  90     listTypeIterator *li = zmalloc(sizeof(listTypeIterator));
  91     li->subject = subject;
  92     li->encoding = subject->encoding;
  93     li->direction = direction;
  94     if (li->encoding == REDIS_ENCODING_ZIPLIST) {
  95         li->zi = ziplistIndex(subject->ptr,index);
  96     } else if (li->encoding == REDIS_ENCODING_LINKEDLIST) {
  97         li->ln = listIndex(subject->ptr,index);
  98     } else {
  99         redisPanic("Unknown list encoding");
 100     }
 101     return li;
 102 }
 103
 104 /* Clean up the iterator. */
 105 void listTypeReleaseIterator(listTypeIterator *li) {
 106     zfree(li);
 107 }
 108
 109 /* Stores pointer to current the entry in the provided entry structure
 110  * and advances the position of the iterator. Returns 1 when the current
 111  * entry is in fact an entry, 0 otherwise. */
 112 int listTypeNext(listTypeIterator *li, listTypeEntry *entry) {
 113     /* Protect from converting when iterating */
 114     redisAssert(li->subject->encoding == li->encoding);
 115
 116     entry->li = li;
 117     if (li->encoding == REDIS_ENCODING_ZIPLIST) {
 118         entry->zi = li->zi;
 119         if (entry->zi != NULL) {
 120             if (li->direction == REDIS_TAIL)
 121                 li->zi = ziplistNext(li->subject->ptr,li->zi);
 122             else
 123                 li->zi = ziplistPrev(li->subject->ptr,li->zi);
 124             return 1;
 125         }
 126     } else if (li->encoding == REDIS_ENCODING_LINKEDLIST) {
 127         entry->ln = li->ln;
 128         if (entry->ln != NULL) {
 129             if (li->direction == REDIS_TAIL)
 130                 li->ln = li->ln->next;
 131             else
 132                 li->ln = li->ln->prev;
 133             return 1;
 134         }
 135     } else {
 136         redisPanic("Unknown list encoding");
 137     }
 138     return 0;
 139 }
 140
 141 /* Return entry or NULL at the current position of the iterator. */
 142 robj *listTypeGet(listTypeEntry *entry) {
 143     listTypeIterator *li = entry->li;
 144     robj *value = NULL;
 145     if (li->encoding == REDIS_ENCODING_ZIPLIST) {
 146         unsigned char *vstr;
 147         unsigned int vlen;
 148         long long vlong;
 149         redisAssert(entry->zi != NULL);
 150         if (ziplistGet(entry->zi,&vstr,&vlen,&vlong)) {
 151             if (vstr) {
 152                 value = createStringObject((char*)vstr,vlen);
 153             } else {
 154                 value = createStringObjectFromLongLong(vlong);
 155             }
 156         }
 157     } else if (li->encoding == REDIS_ENCODING_LINKEDLIST) {
 158         redisAssert(entry->ln != NULL);
 159         value = listNodeValue(entry->ln);
 160         incrRefCount(value);
 161     } else {
 162         redisPanic("Unknown list encoding");
 163     }
 164     return value;
 165 }
 166
 167 void listTypeInsert(listTypeEntry *entry, robj *value, int where) {
 168     robj *subject = entry->li->subject;
 169     if (entry->li->encoding == REDIS_ENCODING_ZIPLIST) {
 170         value = getDecodedObject(value);
 171         if (where == REDIS_TAIL) {
 172             unsigned char *next = ziplistNext(subject->ptr,entry->zi);
 173
 174             /* When we insert after the current element, but the current element
 175              * is the tail of the list, we need to do a push. */
 176             if (next == NULL) {
 177                 subject->ptr = ziplistPush(subject->ptr,value->ptr,sdslen(value->ptr),REDIS_TAIL);
 178             } else {
 179                 subject->ptr = ziplistInsert(subject->ptr,next,value->ptr,sdslen(value->ptr));
 180             }
 181         } else {
 182             subject->ptr = ziplistInsert(subject->ptr,entry->zi,value->ptr,sdslen(value->ptr));
 183         }
 184         decrRefCount(value);
 185     } else if (entry->li->encoding == REDIS_ENCODING_LINKEDLIST) {
 186         if (where == REDIS_TAIL) {
 187             listInsertNode(subject->ptr,entry->ln,value,AL_START_TAIL);
 188         } else {
 189             listInsertNode(subject->ptr,entry->ln,value,AL_START_HEAD);
 190         }
 191         incrRefCount(value);
 192     } else {
 193         redisPanic("Unknown list encoding");
 194     }
 195 }
 196
 197 /* Compare the given object with the entry at the current position. */
 198 int listTypeEqual(listTypeEntry *entry, robj *o) {
 199     listTypeIterator *li = entry->li;
 200     if (li->encoding == REDIS_ENCODING_ZIPLIST) {
 201         redisAssert(o->encoding == REDIS_ENCODING_RAW);
 202         return ziplistCompare(entry->zi,o->ptr,sdslen(o->ptr));
 203     } else if (li->encoding == REDIS_ENCODING_LINKEDLIST) {
 204         return equalStringObjects(o,listNodeValue(entry->ln));
 205     } else {
 206         redisPanic("Unknown list encoding");
 207     }
 208 }
 209
 210 /* Delete the element pointed to. */
 211 void listTypeDelete(listTypeEntry *entry) {
 212     listTypeIterator *li = entry->li;
 213     if (li->encoding == REDIS_ENCODING_ZIPLIST) {
 214         unsigned char *p = entry->zi;
 215         li->subject->ptr = ziplistDelete(li->subject->ptr,&p);
 216
 217         /* Update position of the iterator depending on the direction */
 218         if (li->direction == REDIS_TAIL)
 219             li->zi = p;
 220         else
 221             li->zi = ziplistPrev(li->subject->ptr,p);
 222     } else if (entry->li->encoding == REDIS_ENCODING_LINKEDLIST) {
 223         listNode *next;
 224         if (li->direction == REDIS_TAIL)
 225             next = entry->ln->next;
 226         else
 227             next = entry->ln->prev;
 228         listDelNode(li->subject->ptr,entry->ln);
 229         li->ln = next;
 230     } else {
 231         redisPanic("Unknown list encoding");
 232     }
 233 }
 234
 235 void listTypeConvert(robj *subject, int enc) {
 236     listTypeIterator *li;
 237     listTypeEntry entry;
 238     redisAssert(subject->type == REDIS_LIST);
 239
 240     if (enc == REDIS_ENCODING_LINKEDLIST) {
 241         list *l = listCreate();
 242         listSetFreeMethod(l,decrRefCount);
 243
 244         /* listTypeGet returns a robj with incremented refcount */
 245         li = listTypeInitIterator(subject,0,REDIS_TAIL);
 246         while (listTypeNext(li,&entry)) listAddNodeTail(l,listTypeGet(&entry));
 247         listTypeReleaseIterator(li);
 248
 249         subject->encoding = REDIS_ENCODING_LINKEDLIST;
 250         zfree(subject->ptr);
 251         subject->ptr = l;
 252     } else {
 253         redisPanic("Unsupported list conversion");
 254     }
 255 }
 256
 257 /*-----------------------------------------------------------------------------
 258  * List Commands
 259  *----------------------------------------------------------------------------*/
 260
 261 void pushGenericCommand(redisClient *c, int where) {
 262     robj *lobj = lookupKeyWrite(c->db,c->argv[1]);
 263     c->argv[2] = tryObjectEncoding(c->argv[2]);
 264     if (lobj == NULL) {
 265         if (handleClientsWaitingListPush(c,c->argv[1],c->argv[2])) {
 266             addReply(c,shared.cone);
 267             return;
 268         }
 269         lobj = createZiplistObject();
 270         dbAdd(c->db,c->argv[1],lobj);
 271     } else {
 272         if (lobj->type != REDIS_LIST) {
 273             addReply(c,shared.wrongtypeerr);
 274             return;
 275         }
 276         if (handleClientsWaitingListPush(c,c->argv[1],c->argv[2])) {
 277             touchWatchedKey(c->db,c->argv[1]);
 278             addReply(c,shared.cone);
 279             return;
 280         }
 281     }
 282     listTypePush(lobj,c->argv[2],where);
 283     addReplyLongLong(c,listTypeLength(lobj));
 284     touchWatchedKey(c->db,c->argv[1]);
 285     server.dirty++;
 286 }
 287
 288 void lpushCommand(redisClient *c) {
 289     pushGenericCommand(c,REDIS_HEAD);
 290 }
 291
 292 void rpushCommand(redisClient *c) {
 293     pushGenericCommand(c,REDIS_TAIL);
 294 }
 295
 296 void pushxGenericCommand(redisClient *c, robj *refval, robj *val, int where) {
 297     robj *subject;
 298     listTypeIterator *iter;
 299     listTypeEntry entry;
 300     int inserted = 0;
 301
 302     if ((subject = lookupKeyReadOrReply(c,c->argv[1],shared.czero)) == NULL ||
 303         checkType(c,subject,REDIS_LIST)) return;
 304
 305     if (refval != NULL) {
 306         /* Note: we expect refval to be string-encoded because it is *not* the
 307          * last argument of the multi-bulk LINSERT. */
 308         redisAssert(refval->encoding == REDIS_ENCODING_RAW);
 309
 310         /* We're not sure if this value can be inserted yet, but we cannot
 311          * convert the list inside the iterator. We don't want to loop over
 312          * the list twice (once to see if the value can be inserted and once
 313          * to do the actual insert), so we assume this value can be inserted
 314          * and convert the ziplist to a regular list if necessary. */
 315         listTypeTryConversion(subject,val);
 316
 317         /* Seek refval from head to tail */
 318         iter = listTypeInitIterator(subject,0,REDIS_TAIL);
 319         while (listTypeNext(iter,&entry)) {
 320             if (listTypeEqual(&entry,refval)) {
 321                 listTypeInsert(&entry,val,where);
 322                 inserted = 1;
 323                 break;
 324             }
 325         }
 326         listTypeReleaseIterator(iter);
 327
 328         if (inserted) {
 329             /* Check if the length exceeds the ziplist length threshold. */
 330             if (subject->encoding == REDIS_ENCODING_ZIPLIST &&
 331                 ziplistLen(subject->ptr) > server.list_max_ziplist_entries)
 332                     listTypeConvert(subject,REDIS_ENCODING_LINKEDLIST);
 333             touchWatchedKey(c->db,c->argv[1]);
 334             server.dirty++;
 335         } else {
 336             /* Notify client of a failed insert */
 337             addReply(c,shared.cnegone);
 338             return;
 339         }
 340     } else {
 341         listTypePush(subject,val,where);
 342         touchWatchedKey(c->db,c->argv[1]);
 343         server.dirty++;
 344     }
 345
 346     addReplyLongLong(c,listTypeLength(subject));
 347 }
 348
 349 void lpushxCommand(redisClient *c) {
 350     c->argv[2] = tryObjectEncoding(c->argv[2]);
 351     pushxGenericCommand(c,NULL,c->argv[2],REDIS_HEAD);
 352 }
 353
 354 void rpushxCommand(redisClient *c) {
 355     c->argv[2] = tryObjectEncoding(c->argv[2]);
 356     pushxGenericCommand(c,NULL,c->argv[2],REDIS_TAIL);
 357 }
 358
 359 void linsertCommand(redisClient *c) {
 360     c->argv[4] = tryObjectEncoding(c->argv[4]);
 361     if (strcasecmp(c->argv[2]->ptr,"after") == 0) {
 362         pushxGenericCommand(c,c->argv[3],c->argv[4],REDIS_TAIL);
 363     } else if (strcasecmp(c->argv[2]->ptr,"before") == 0) {
 364         pushxGenericCommand(c,c->argv[3],c->argv[4],REDIS_HEAD);
 365     } else {
 366         addReply(c,shared.syntaxerr);
 367     }
 368 }
 369
 370 void llenCommand(redisClient *c) {
 371     robj *o = lookupKeyReadOrReply(c,c->argv[1],shared.czero);
 372     if (o == NULL || checkType(c,o,REDIS_LIST)) return;
 373     addReplyLongLong(c,listTypeLength(o));
 374 }
 375
 376 void lindexCommand(redisClient *c) {
 377     robj *o = lookupKeyReadOrReply(c,c->argv[1],shared.nullbulk);
 378     if (o == NULL || checkType(c,o,REDIS_LIST)) return;
 379     int index = atoi(c->argv[2]->ptr);
 380     robj *value = NULL;
 381
 382     if (o->encoding == REDIS_ENCODING_ZIPLIST) {
 383         unsigned char *p;
 384         unsigned char *vstr;
 385         unsigned int vlen;
 386         long long vlong;
 387         p = ziplistIndex(o->ptr,index);
 388         if (ziplistGet(p,&vstr,&vlen,&vlong)) {
 389             if (vstr) {
 390                 value = createStringObject((char*)vstr,vlen);
 391             } else {
 392                 value = createStringObjectFromLongLong(vlong);
 393             }
 394             addReplyBulk(c,value);
 395             decrRefCount(value);
 396         } else {
 397             addReply(c,shared.nullbulk);
 398         }
 399     } else if (o->encoding == REDIS_ENCODING_LINKEDLIST) {
 400         listNode *ln = listIndex(o->ptr,index);
 401         if (ln != NULL) {
 402             value = listNodeValue(ln);
 403             addReplyBulk(c,value);
 404         } else {
 405             addReply(c,shared.nullbulk);
 406         }
 407     } else {
 408         redisPanic("Unknown list encoding");
 409     }
 410 }
 411
 412 void lsetCommand(redisClient *c) {
 413     robj *o = lookupKeyWriteOrReply(c,c->argv[1],shared.nokeyerr);
 414     if (o == NULL || checkType(c,o,REDIS_LIST)) return;
 415     int index = atoi(c->argv[2]->ptr);
 416     robj *value = (c->argv[3] = tryObjectEncoding(c->argv[3]));
 417
 418     listTypeTryConversion(o,value);
 419     if (o->encoding == REDIS_ENCODING_ZIPLIST) {
 420         unsigned char *p, *zl = o->ptr;
 421         p = ziplistIndex(zl,index);
 422         if (p == NULL) {
 423             addReply(c,shared.outofrangeerr);
 424         } else {
 425             o->ptr = ziplistDelete(o->ptr,&p);
 426             value = getDecodedObject(value);
 427             o->ptr = ziplistInsert(o->ptr,p,value->ptr,sdslen(value->ptr));
 428             decrRefCount(value);
 429             addReply(c,shared.ok);
 430             touchWatchedKey(c->db,c->argv[1]);
 431             server.dirty++;
 432         }
 433     } else if (o->encoding == REDIS_ENCODING_LINKEDLIST) {
 434         listNode *ln = listIndex(o->ptr,index);
 435         if (ln == NULL) {
 436             addReply(c,shared.outofrangeerr);
 437         } else {
 438             decrRefCount((robj*)listNodeValue(ln));
 439             listNodeValue(ln) = value;
 440             incrRefCount(value);
 441             addReply(c,shared.ok);
 442             touchWatchedKey(c->db,c->argv[1]);
 443             server.dirty++;
 444         }
 445     } else {
 446         redisPanic("Unknown list encoding");
 447     }
 448 }
 449
 450 void popGenericCommand(redisClient *c, int where) {
 451     robj *o = lookupKeyWriteOrReply(c,c->argv[1],shared.nullbulk);
 452     if (o == NULL || checkType(c,o,REDIS_LIST)) return;
 453
 454     robj *value = listTypePop(o,where);
 455     if (value == NULL) {
 456         addReply(c,shared.nullbulk);
 457     } else {
 458         addReplyBulk(c,value);
 459         decrRefCount(value);
 460         if (listTypeLength(o) == 0) dbDelete(c->db,c->argv[1]);
 461         touchWatchedKey(c->db,c->argv[1]);
 462         server.dirty++;
 463     }
 464 }
 465
 466 void lpopCommand(redisClient *c) {
 467     popGenericCommand(c,REDIS_HEAD);
 468 }
 469
 470 void rpopCommand(redisClient *c) {
 471     popGenericCommand(c,REDIS_TAIL);
 472 }
 473
 474 void lrangeCommand(redisClient *c) {
 475     robj *o, *value;
 476     int start = atoi(c->argv[2]->ptr);
 477     int end = atoi(c->argv[3]->ptr);
 478     int llen;
 479     int rangelen, j;
 480     listTypeEntry entry;
 481
 482     if ((o = lookupKeyReadOrReply(c,c->argv[1],shared.emptymultibulk)) == NULL
 483          || checkType(c,o,REDIS_LIST)) return;
 484     llen = listTypeLength(o);
 485
 486     /* convert negative indexes */
 487     if (start < 0) start = llen+start;
 488     if (end < 0) end = llen+end;
 489     if (start < 0) start = 0;
 490
 491     /* Invariant: start >= 0, so this test will be true when end < 0.
 492      * The range is empty when start > end or start >= length. */
 493     if (start > end || start >= llen) {
 494         addReply(c,shared.emptymultibulk);
 495         return;
 496     }
 497     if (end >= llen) end = llen-1;
 498     rangelen = (end-start)+1;
 499
 500     /* Return the result in form of a multi-bulk reply */
 501     addReplyMultiBulkLen(c,rangelen);
 502     listTypeIterator *li = listTypeInitIterator(o,start,REDIS_TAIL);
 503     for (j = 0; j < rangelen; j++) {
 504         redisAssert(listTypeNext(li,&entry));
 505         value = listTypeGet(&entry);
 506         addReplyBulk(c,value);
 507         decrRefCount(value);
 508     }
 509     listTypeReleaseIterator(li);
 510 }
 511
 512 void ltrimCommand(redisClient *c) {
 513     robj *o;
 514     int start = atoi(c->argv[2]->ptr);
 515     int end = atoi(c->argv[3]->ptr);
 516     int llen;
 517     int j, ltrim, rtrim;
 518     list *list;
 519     listNode *ln;
 520
 521     if ((o = lookupKeyWriteOrReply(c,c->argv[1],shared.ok)) == NULL ||
 522         checkType(c,o,REDIS_LIST)) return;
 523     llen = listTypeLength(o);
 524
 525     /* convert negative indexes */
 526     if (start < 0) start = llen+start;
 527     if (end < 0) end = llen+end;
 528     if (start < 0) start = 0;
 529
 530     /* Invariant: start >= 0, so this test will be true when end < 0.
 531      * The range is empty when start > end or start >= length. */
 532     if (start > end || start >= llen) {
 533         /* Out of range start or start > end result in empty list */
 534         ltrim = llen;
 535         rtrim = 0;
 536     } else {
 537         if (end >= llen) end = llen-1;
 538         ltrim = start;
 539         rtrim = llen-end-1;
 540     }
 541
 542     /* Remove list elements to perform the trim */
 543     if (o->encoding == REDIS_ENCODING_ZIPLIST) {
 544         o->ptr = ziplistDeleteRange(o->ptr,0,ltrim);
 545         o->ptr = ziplistDeleteRange(o->ptr,-rtrim,rtrim);
 546     } else if (o->encoding == REDIS_ENCODING_LINKEDLIST) {
 547         list = o->ptr;
 548         for (j = 0; j < ltrim; j++) {
 549             ln = listFirst(list);
 550             listDelNode(list,ln);
 551         }
 552         for (j = 0; j < rtrim; j++) {
 553             ln = listLast(list);
 554             listDelNode(list,ln);
 555         }
 556     } else {
 557         redisPanic("Unknown list encoding");
 558     }
 559     if (listTypeLength(o) == 0) dbDelete(c->db,c->argv[1]);
 560     touchWatchedKey(c->db,c->argv[1]);
 561     server.dirty++;
 562     addReply(c,shared.ok);
 563 }
 564
 565 void lremCommand(redisClient *c) {
 566     robj *subject, *obj;
 567     obj = c->argv[3] = tryObjectEncoding(c->argv[3]);
 568     int toremove = atoi(c->argv[2]->ptr);
 569     int removed = 0;
 570     listTypeEntry entry;
 571
 572     subject = lookupKeyWriteOrReply(c,c->argv[1],shared.czero);
 573     if (subject == NULL || checkType(c,subject,REDIS_LIST)) return;
 574
 575     /* Make sure obj is raw when we're dealing with a ziplist */
 576     if (subject->encoding == REDIS_ENCODING_ZIPLIST)
 577         obj = getDecodedObject(obj);
 578
 579     listTypeIterator *li;
 580     if (toremove < 0) {
 581         toremove = -toremove;
 582         li = listTypeInitIterator(subject,-1,REDIS_HEAD);
 583     } else {
 584         li = listTypeInitIterator(subject,0,REDIS_TAIL);
 585     }
 586
 587     while (listTypeNext(li,&entry)) {
 588         if (listTypeEqual(&entry,obj)) {
 589             listTypeDelete(&entry);
 590             server.dirty++;
 591             removed++;
 592             if (toremove && removed == toremove) break;
 593         }
 594     }
 595     listTypeReleaseIterator(li);
 596
 597     /* Clean up raw encoded object */
 598     if (subject->encoding == REDIS_ENCODING_ZIPLIST)
 599         decrRefCount(obj);
 600
 601     if (listTypeLength(subject) == 0) dbDelete(c->db,c->argv[1]);
 602     addReplyLongLong(c,removed);
 603     if (removed) touchWatchedKey(c->db,c->argv[1]);
 604 }
 605
 606 /* This is the semantic of this command:
 607  *  RPOPLPUSH srclist dstlist:
 608  *    IF LLEN(srclist) > 0
 609  *      element = RPOP srclist
 610  *      LPUSH dstlist element
 611  *      RETURN element
 612  *    ELSE
 613  *      RETURN nil
 614  *    END
 615  *  END
 616  *
 617  * The idea is to be able to get an element from a list in a reliable way
 618  * since the element is not just returned but pushed against another list
 619  * as well. This command was originally proposed by Ezra Zygmuntowicz.
 620  */
 621
 622 void rpoplpushHandlePush(redisClient *c, robj *dstkey, robj *dstobj, robj *value) {
 623     if (!handleClientsWaitingListPush(c,dstkey,value)) {
 624         /* Create the list if the key does not exist */
 625         if (!dstobj) {
 626             dstobj = createZiplistObject();
 627             dbAdd(c->db,dstkey,dstobj);
 628         } else {
 629             touchWatchedKey(c->db,dstkey);
 630             server.dirty++;
 631         }
 632         listTypePush(dstobj,value,REDIS_HEAD);
 633     }
 634
 635     /* Always send the pushed value to the client. */
 636     addReplyBulk(c,value);
 637 }
 638
 639 void rpoplpushCommand(redisClient *c) {
 640     robj *sobj, *value;
 641     if ((sobj = lookupKeyWriteOrReply(c,c->argv[1],shared.nullbulk)) == NULL ||
 642         checkType(c,sobj,REDIS_LIST)) return;
 643
 644     if (listTypeLength(sobj) == 0) {
 645         addReply(c,shared.nullbulk);
 646     } else {
 647         robj *dobj = lookupKeyWrite(c->db,c->argv[2]);
 648         if (dobj && checkType(c,dobj,REDIS_LIST)) return;
 649         value = listTypePop(sobj,REDIS_TAIL);
 650         rpoplpushHandlePush(c,c->argv[2],dobj,value);
 651
 652         /* listTypePop returns an object with its refcount incremented */
 653         decrRefCount(value);
 654
 655         /* Delete the source list when it is empty */
 656         if (listTypeLength(sobj) == 0) dbDelete(c->db,c->argv[1]);
 657         touchWatchedKey(c->db,c->argv[1]);
 658         server.dirty++;
 659     }
 660 }
 661
 662 /*-----------------------------------------------------------------------------
 663  * Blocking POP operations
 664  *----------------------------------------------------------------------------*/
 665
 666 /* Currently Redis blocking operations support is limited to list POP ops,
 667  * so the current implementation is not fully generic, but it is also not
 668  * completely specific so it will not require a rewrite to support new
 669  * kind of blocking operations in the future.
 670  *
 671  * Still it's important to note that list blocking operations can be already
 672  * used as a notification mechanism in order to implement other blocking
 673  * operations at application level, so there must be a very strong evidence
 674  * of usefulness and generality before new blocking operations are implemented.
 675  *
 676  * This is how the current blocking POP works, we use BLPOP as example:
 677  * - If the user calls BLPOP and the key exists and contains a non empty list
 678  *   then LPOP is called instead. So BLPOP is semantically the same as LPOP
 679  *   if there is not to block.
 680  * - If instead BLPOP is called and the key does not exists or the list is
 681  *   empty we need to block. In order to do so we remove the notification for
 682  *   new data to read in the client socket (so that we'll not serve new
 683  *   requests if the blocking request is not served). Also we put the client
 684  *   in a dictionary (db->blocking_keys) mapping keys to a list of clients
 685  *   blocking for this keys.
 686  * - If a PUSH operation against a key with blocked clients waiting is
 687  *   performed, we serve the first in the list: basically instead to push
 688  *   the new element inside the list we return it to the (first / oldest)
 689  *   blocking client, unblock the client, and remove it form the list.
 690  *
 691  * The above comment and the source code should be enough in order to understand
 692  * the implementation and modify / fix it later.
 693  */
 694
 695 /* Set a client in blocking mode for the specified key, with the specified
 696  * timeout */
 697 void blockForKeys(redisClient *c, robj **keys, int numkeys, time_t timeout, robj *target) {
 698     dictEntry *de;
 699     list *l;
 700     int j;
 701
 702     c->bpop.keys = zmalloc(sizeof(robj*)*numkeys);
 703     c->bpop.count = numkeys;
 704     c->bpop.timeout = timeout;
 705     c->bpop.target = target;
 706
 707     if (target != NULL) {
 708       incrRefCount(target);
 709     }
 710
 711     for (j = 0; j < numkeys; j++) {
 712         /* Add the key in the client structure, to map clients -> keys */
 713         c->bpop.keys[j] = keys[j];
 714         incrRefCount(keys[j]);
 715
 716         /* And in the other "side", to map keys -> clients */
 717         de = dictFind(c->db->blocking_keys,keys[j]);
 718         if (de == NULL) {
 719             int retval;
 720
 721             /* For every key we take a list of clients blocked for it */
 722             l = listCreate();
 723             retval = dictAdd(c->db->blocking_keys,keys[j],l);
 724             incrRefCount(keys[j]);
 725             redisAssert(retval == DICT_OK);
 726         } else {
 727             l = dictGetEntryVal(de);
 728         }
 729         listAddNodeTail(l,c);
 730     }
 731     /* Mark the client as a blocked client */
 732     c->flags |= REDIS_BLOCKED;
 733     server.bpop_blocked_clients++;
 734 }
 735
 736 /* Unblock a client that's waiting in a blocking operation such as BLPOP */
 737 void unblockClientWaitingData(redisClient *c) {
 738     dictEntry *de;
 739     list *l;
 740     int j;
 741
 742     redisAssert(c->bpop.keys != NULL);
 743     /* The client may wait for multiple keys, so unblock it for every key. */
 744     for (j = 0; j < c->bpop.count; j++) {
 745         /* Remove this client from the list of clients waiting for this key. */
 746         de = dictFind(c->db->blocking_keys,c->bpop.keys[j]);
 747         redisAssert(de != NULL);
 748         l = dictGetEntryVal(de);
 749         listDelNode(l,listSearchKey(l,c));
 750         /* If the list is empty we need to remove it to avoid wasting memory */
 751         if (listLength(l) == 0)
 752             dictDelete(c->db->blocking_keys,c->bpop.keys[j]);
 753         decrRefCount(c->bpop.keys[j]);
 754     }
 755
 756     /* Cleanup the client structure */
 757     zfree(c->bpop.keys);
 758     c->bpop.keys = NULL;
 759     c->bpop.target = NULL;
 760     c->flags &= (~REDIS_BLOCKED);
 761     server.bpop_blocked_clients--;
 762     /* We want to process data if there is some command waiting
 763      * in the input buffer. Note that this is safe even if
 764      * unblockClientWaitingData() gets called from freeClient() because
 765      * freeClient() will be smart enough to call this function
 766      * *after* c->querybuf was set to NULL. */
 767     if (c->querybuf && sdslen(c->querybuf) > 0) processInputBuffer(c);
 768 }
 769
 770 /* This should be called from any function PUSHing into lists.
 771  * 'c' is the "pushing client", 'key' is the key it is pushing data against,
 772  * 'ele' is the element pushed.
 773  *
 774  * If the function returns 0 there was no client waiting for a list push
 775  * against this key.
 776  *
 777  * If the function returns 1 there was a client waiting for a list push
 778  * against this key, the element was passed to this client thus it's not
 779  * needed to actually add it to the list and the caller should return asap. */
 780 int handleClientsWaitingListPush(redisClient *c, robj *key, robj *ele) {
 781     struct dictEntry *de;
 782     redisClient *receiver;
 783     int numclients;
 784     list *clients;
 785     listNode *ln;
 786     robj *dstkey, *dstobj;
 787
 788     de = dictFind(c->db->blocking_keys,key);
 789     if (de == NULL) return 0;
 790     clients = dictGetEntryVal(de);
 791     numclients = listLength(clients);
 792
 793     /* Try to handle the push as long as there are clients waiting for a push.
 794      * Note that "numclients" is used because the list of clients waiting for a
 795      * push on "key" is deleted by unblockClient() when empty.
 796      *
 797      * This loop will have more than 1 iteration when there is a BRPOPLPUSH
 798      * that cannot push the target list because it does not contain a list. If
 799      * this happens, it simply tries the next client waiting for a push. */
 800     while (numclients--) {
 801         ln = listFirst(clients);
 802         redisAssert(ln != NULL);
 803         receiver = ln->value;
 804         dstkey = receiver->bpop.target;
 805
 806         /* This should remove the first element of the "clients" list. */
 807         unblockClientWaitingData(receiver);
 808         redisAssert(ln != listFirst(clients));
 809
 810         if (dstkey == NULL) {
 811             /* BRPOP/BLPOP */
 812             addReplyMultiBulkLen(receiver,2);
 813             addReplyBulk(receiver,key);
 814             addReplyBulk(receiver,ele);
 815             return 1;
 816         } else {
 817             /* BRPOPLPUSH */
 818             dstobj = lookupKeyWrite(receiver->db,dstkey);
 819             if (dstobj && checkType(receiver,dstobj,REDIS_LIST)) {
 820                 decrRefCount(dstkey);
 821             } else {
 822                 rpoplpushHandlePush(receiver,dstkey,dstobj,ele);
 823                 decrRefCount(dstkey);
 824                 return 1;
 825             }
 826         }
 827     }
 828
 829     return 0;
 830 }
 831
 832 int getTimeoutFromObjectOrReply(redisClient *c, robj *object, time_t *timeout) {
 833     long tval;
 834
 835     if (getLongFromObjectOrReply(c,object,&tval,
 836         "timeout is not an integer or out of range") != REDIS_OK)
 837         return REDIS_ERR;
 838
 839     if (tval < 0) {
 840         addReplyError(c,"timeout is negative");
 841         return REDIS_ERR;
 842     }
 843
 844     if (tval > 0) tval += time(NULL);
 845     *timeout = tval;
 846
 847     return REDIS_OK;
 848 }
 849
 850 /* Blocking RPOP/LPOP */
 851 void blockingPopGenericCommand(redisClient *c, int where) {
 852     robj *o;
 853     time_t timeout;
 854     int j;
 855
 856     if (getTimeoutFromObjectOrReply(c,c->argv[c->argc-1],&timeout) != REDIS_OK)
 857         return;
 858
 859     for (j = 1; j < c->argc-1; j++) {
 860         o = lookupKeyWrite(c->db,c->argv[j]);
 861         if (o != NULL) {
 862             if (o->type != REDIS_LIST) {
 863                 addReply(c,shared.wrongtypeerr);
 864                 return;
 865             } else {
 866                 if (listTypeLength(o) != 0) {
 867                     /* If the list contains elements fall back to the usual
 868                      * non-blocking POP operation */
 869                     robj *argv[2], **orig_argv;
 870                     int orig_argc;
 871
 872                     /* We need to alter the command arguments before to call
 873                      * popGenericCommand() as the command takes a single key. */
 874                     orig_argv = c->argv;
 875                     orig_argc = c->argc;
 876                     argv[1] = c->argv[j];
 877                     c->argv = argv;
 878                     c->argc = 2;
 879
 880                     /* Also the return value is different, we need to output
 881                      * the multi bulk reply header and the key name. The
 882                      * "real" command will add the last element (the value)
 883                      * for us. If this souds like an hack to you it's just
 884                      * because it is... */
 885                     addReplyMultiBulkLen(c,2);
 886                     addReplyBulk(c,argv[1]);
 887
 888                     popGenericCommand(c,where);
 889
 890                     /* Fix the client structure with the original stuff */
 891                     c->argv = orig_argv;
 892                     c->argc = orig_argc;
 893
 894                     return;
 895                 }
 896             }
 897         }
 898     }
 899
 900     /* If we are inside a MULTI/EXEC and the list is empty the only thing
 901      * we can do is treating it as a timeout (even with timeout 0). */
 902     if (c->flags & REDIS_MULTI) {
 903         addReply(c,shared.nullmultibulk);
 904         return;
 905     }
 906
 907     /* If the list is empty or the key does not exists we must block */
 908     blockForKeys(c, c->argv + 1, c->argc - 2, timeout, NULL);
 909 }
 910
 911 void blpopCommand(redisClient *c) {
 912     blockingPopGenericCommand(c,REDIS_HEAD);
 913 }
 914
 915 void brpopCommand(redisClient *c) {
 916     blockingPopGenericCommand(c,REDIS_TAIL);
 917 }
 918
 919 void brpoplpushCommand(redisClient *c) {
 920     time_t timeout;
 921
 922     if (getTimeoutFromObjectOrReply(c,c->argv[3],&timeout) != REDIS_OK)
 923         return;
 924
 925     robj *key = lookupKeyWrite(c->db, c->argv[1]);
 926
 927     if (key == NULL) {
 928         if (c->flags & REDIS_MULTI) {
 929
 930             /* Blocking against an empty list in a multi state
 931              * returns immediately. */
 932             addReply(c, shared.nullmultibulk);
 933         } else {
 934             /* The list is empty and the client blocks. */
 935             blockForKeys(c, c->argv + 1, 1, timeout, c->argv[2]);
 936         }
 937     } else {
 938         if (key->type != REDIS_LIST) {
 939             addReply(c, shared.wrongtypeerr);
 940         } else {
 941
 942             /* The list exists and has elements, so
 943              * the regular rpoplpushCommand is executed. */
 944             redisAssert(listTypeLength(key) > 0);
 945             rpoplpushCommand(c);
 946         }
 947     }
 948 }