t_list.c 34.2 KB
Newer Older
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88
#include "redis.h"

/*-----------------------------------------------------------------------------
 * List API
 *----------------------------------------------------------------------------*/

/* Check the argument length to see if it requires us to convert the ziplist
 * to a real list. Only check raw-encoded objects because integer encoded
 * objects are never too long. */
void listTypeTryConversion(robj *subject, robj *value) {
    if (subject->encoding != REDIS_ENCODING_ZIPLIST) return;
    if (value->encoding == REDIS_ENCODING_RAW &&
        sdslen(value->ptr) > server.list_max_ziplist_value)
            listTypeConvert(subject,REDIS_ENCODING_LINKEDLIST);
}

void listTypePush(robj *subject, robj *value, int where) {
    /* Check if we need to convert the ziplist */
    listTypeTryConversion(subject,value);
    if (subject->encoding == REDIS_ENCODING_ZIPLIST &&
        ziplistLen(subject->ptr) >= server.list_max_ziplist_entries)
            listTypeConvert(subject,REDIS_ENCODING_LINKEDLIST);

    if (subject->encoding == REDIS_ENCODING_ZIPLIST) {
        int pos = (where == REDIS_HEAD) ? ZIPLIST_HEAD : ZIPLIST_TAIL;
        value = getDecodedObject(value);
        subject->ptr = ziplistPush(subject->ptr,value->ptr,sdslen(value->ptr),pos);
        decrRefCount(value);
    } else if (subject->encoding == REDIS_ENCODING_LINKEDLIST) {
        if (where == REDIS_HEAD) {
            listAddNodeHead(subject->ptr,value);
        } else {
            listAddNodeTail(subject->ptr,value);
        }
        incrRefCount(value);
    } else {
        redisPanic("Unknown list encoding");
    }
}

robj *listTypePop(robj *subject, int where) {
    robj *value = NULL;
    if (subject->encoding == REDIS_ENCODING_ZIPLIST) {
        unsigned char *p;
        unsigned char *vstr;
        unsigned int vlen;
        long long vlong;
        int pos = (where == REDIS_HEAD) ? 0 : -1;
        p = ziplistIndex(subject->ptr,pos);
        if (ziplistGet(p,&vstr,&vlen,&vlong)) {
            if (vstr) {
                value = createStringObject((char*)vstr,vlen);
            } else {
                value = createStringObjectFromLongLong(vlong);
            }
            /* We only need to delete an element when it exists */
            subject->ptr = ziplistDelete(subject->ptr,&p);
        }
    } else if (subject->encoding == REDIS_ENCODING_LINKEDLIST) {
        list *list = subject->ptr;
        listNode *ln;
        if (where == REDIS_HEAD) {
            ln = listFirst(list);
        } else {
            ln = listLast(list);
        }
        if (ln != NULL) {
            value = listNodeValue(ln);
            incrRefCount(value);
            listDelNode(list,ln);
        }
    } else {
        redisPanic("Unknown list encoding");
    }
    return value;
}

unsigned long listTypeLength(robj *subject) {
    if (subject->encoding == REDIS_ENCODING_ZIPLIST) {
        return ziplistLen(subject->ptr);
    } else if (subject->encoding == REDIS_ENCODING_LINKEDLIST) {
        return listLength((list*)subject->ptr);
    } else {
        redisPanic("Unknown list encoding");
    }
}

/* Initialize an iterator at the specified index. */
89
listTypeIterator *listTypeInitIterator(robj *subject, long index, unsigned char direction) {
90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200
    listTypeIterator *li = zmalloc(sizeof(listTypeIterator));
    li->subject = subject;
    li->encoding = subject->encoding;
    li->direction = direction;
    if (li->encoding == REDIS_ENCODING_ZIPLIST) {
        li->zi = ziplistIndex(subject->ptr,index);
    } else if (li->encoding == REDIS_ENCODING_LINKEDLIST) {
        li->ln = listIndex(subject->ptr,index);
    } else {
        redisPanic("Unknown list encoding");
    }
    return li;
}

/* Clean up the iterator. */
void listTypeReleaseIterator(listTypeIterator *li) {
    zfree(li);
}

/* Stores pointer to current the entry in the provided entry structure
 * and advances the position of the iterator. Returns 1 when the current
 * entry is in fact an entry, 0 otherwise. */
int listTypeNext(listTypeIterator *li, listTypeEntry *entry) {
    /* Protect from converting when iterating */
    redisAssert(li->subject->encoding == li->encoding);

    entry->li = li;
    if (li->encoding == REDIS_ENCODING_ZIPLIST) {
        entry->zi = li->zi;
        if (entry->zi != NULL) {
            if (li->direction == REDIS_TAIL)
                li->zi = ziplistNext(li->subject->ptr,li->zi);
            else
                li->zi = ziplistPrev(li->subject->ptr,li->zi);
            return 1;
        }
    } else if (li->encoding == REDIS_ENCODING_LINKEDLIST) {
        entry->ln = li->ln;
        if (entry->ln != NULL) {
            if (li->direction == REDIS_TAIL)
                li->ln = li->ln->next;
            else
                li->ln = li->ln->prev;
            return 1;
        }
    } else {
        redisPanic("Unknown list encoding");
    }
    return 0;
}

/* Return entry or NULL at the current position of the iterator. */
robj *listTypeGet(listTypeEntry *entry) {
    listTypeIterator *li = entry->li;
    robj *value = NULL;
    if (li->encoding == REDIS_ENCODING_ZIPLIST) {
        unsigned char *vstr;
        unsigned int vlen;
        long long vlong;
        redisAssert(entry->zi != NULL);
        if (ziplistGet(entry->zi,&vstr,&vlen,&vlong)) {
            if (vstr) {
                value = createStringObject((char*)vstr,vlen);
            } else {
                value = createStringObjectFromLongLong(vlong);
            }
        }
    } else if (li->encoding == REDIS_ENCODING_LINKEDLIST) {
        redisAssert(entry->ln != NULL);
        value = listNodeValue(entry->ln);
        incrRefCount(value);
    } else {
        redisPanic("Unknown list encoding");
    }
    return value;
}

void listTypeInsert(listTypeEntry *entry, robj *value, int where) {
    robj *subject = entry->li->subject;
    if (entry->li->encoding == REDIS_ENCODING_ZIPLIST) {
        value = getDecodedObject(value);
        if (where == REDIS_TAIL) {
            unsigned char *next = ziplistNext(subject->ptr,entry->zi);

            /* When we insert after the current element, but the current element
             * is the tail of the list, we need to do a push. */
            if (next == NULL) {
                subject->ptr = ziplistPush(subject->ptr,value->ptr,sdslen(value->ptr),REDIS_TAIL);
            } else {
                subject->ptr = ziplistInsert(subject->ptr,next,value->ptr,sdslen(value->ptr));
            }
        } else {
            subject->ptr = ziplistInsert(subject->ptr,entry->zi,value->ptr,sdslen(value->ptr));
        }
        decrRefCount(value);
    } else if (entry->li->encoding == REDIS_ENCODING_LINKEDLIST) {
        if (where == REDIS_TAIL) {
            listInsertNode(subject->ptr,entry->ln,value,AL_START_TAIL);
        } else {
            listInsertNode(subject->ptr,entry->ln,value,AL_START_HEAD);
        }
        incrRefCount(value);
    } else {
        redisPanic("Unknown list encoding");
    }
}

/* Compare the given object with the entry at the current position. */
int listTypeEqual(listTypeEntry *entry, robj *o) {
    listTypeIterator *li = entry->li;
    if (li->encoding == REDIS_ENCODING_ZIPLIST) {
201
        redisAssertWithInfo(NULL,o,o->encoding == REDIS_ENCODING_RAW);
202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237
        return ziplistCompare(entry->zi,o->ptr,sdslen(o->ptr));
    } else if (li->encoding == REDIS_ENCODING_LINKEDLIST) {
        return equalStringObjects(o,listNodeValue(entry->ln));
    } else {
        redisPanic("Unknown list encoding");
    }
}

/* Delete the element pointed to. */
void listTypeDelete(listTypeEntry *entry) {
    listTypeIterator *li = entry->li;
    if (li->encoding == REDIS_ENCODING_ZIPLIST) {
        unsigned char *p = entry->zi;
        li->subject->ptr = ziplistDelete(li->subject->ptr,&p);

        /* Update position of the iterator depending on the direction */
        if (li->direction == REDIS_TAIL)
            li->zi = p;
        else
            li->zi = ziplistPrev(li->subject->ptr,p);
    } else if (entry->li->encoding == REDIS_ENCODING_LINKEDLIST) {
        listNode *next;
        if (li->direction == REDIS_TAIL)
            next = entry->ln->next;
        else
            next = entry->ln->prev;
        listDelNode(li->subject->ptr,entry->ln);
        li->ln = next;
    } else {
        redisPanic("Unknown list encoding");
    }
}

void listTypeConvert(robj *subject, int enc) {
    listTypeIterator *li;
    listTypeEntry entry;
238
    redisAssertWithInfo(NULL,subject,subject->type == REDIS_LIST);
239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261

    if (enc == REDIS_ENCODING_LINKEDLIST) {
        list *l = listCreate();
        listSetFreeMethod(l,decrRefCount);

        /* listTypeGet returns a robj with incremented refcount */
        li = listTypeInitIterator(subject,0,REDIS_TAIL);
        while (listTypeNext(li,&entry)) listAddNodeTail(l,listTypeGet(&entry));
        listTypeReleaseIterator(li);

        subject->encoding = REDIS_ENCODING_LINKEDLIST;
        zfree(subject->ptr);
        subject->ptr = l;
    } else {
        redisPanic("Unsupported list conversion");
    }
}

/*-----------------------------------------------------------------------------
 * List Commands
 *----------------------------------------------------------------------------*/

void pushGenericCommand(redisClient *c, int where) {
A
antirez 已提交
262
    int j, addlen = 0, pushed = 0;
263
    robj *lobj = lookupKeyWrite(c->db,c->argv[1]);
A
antirez 已提交
264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279
    int may_have_waiting_clients = (lobj == NULL);

    if (lobj && lobj->type != REDIS_LIST) {
        addReply(c,shared.wrongtypeerr);
        return;
    }

    for (j = 2; j < c->argc; j++) {
        c->argv[j] = tryObjectEncoding(c->argv[j]);
        if (may_have_waiting_clients) {
            if (handleClientsWaitingListPush(c,c->argv[1],c->argv[j])) {
                addlen++;
                continue;
            } else {
                may_have_waiting_clients = 0;
            }
280
        }
A
antirez 已提交
281 282 283
        if (!lobj) {
            lobj = createZiplistObject();
            dbAdd(c->db,c->argv[1],lobj);
284
        }
A
antirez 已提交
285 286
        listTypePush(lobj,c->argv[j],where);
        pushed++;
287
    }
A
antirez 已提交
288 289 290
    addReplyLongLong(c,addlen + (lobj ? listTypeLength(lobj) : 0));
    if (pushed) signalModifiedKey(c->db,c->argv[1]);
    server.dirty += pushed;
291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312
}

void lpushCommand(redisClient *c) {
    pushGenericCommand(c,REDIS_HEAD);
}

void rpushCommand(redisClient *c) {
    pushGenericCommand(c,REDIS_TAIL);
}

void pushxGenericCommand(redisClient *c, robj *refval, robj *val, int where) {
    robj *subject;
    listTypeIterator *iter;
    listTypeEntry entry;
    int inserted = 0;

    if ((subject = lookupKeyReadOrReply(c,c->argv[1],shared.czero)) == NULL ||
        checkType(c,subject,REDIS_LIST)) return;

    if (refval != NULL) {
        /* Note: we expect refval to be string-encoded because it is *not* the
         * last argument of the multi-bulk LINSERT. */
313
        redisAssertWithInfo(c,refval,refval->encoding == REDIS_ENCODING_RAW);
314 315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337

        /* We're not sure if this value can be inserted yet, but we cannot
         * convert the list inside the iterator. We don't want to loop over
         * the list twice (once to see if the value can be inserted and once
         * to do the actual insert), so we assume this value can be inserted
         * and convert the ziplist to a regular list if necessary. */
        listTypeTryConversion(subject,val);

        /* Seek refval from head to tail */
        iter = listTypeInitIterator(subject,0,REDIS_TAIL);
        while (listTypeNext(iter,&entry)) {
            if (listTypeEqual(&entry,refval)) {
                listTypeInsert(&entry,val,where);
                inserted = 1;
                break;
            }
        }
        listTypeReleaseIterator(iter);

        if (inserted) {
            /* Check if the length exceeds the ziplist length threshold. */
            if (subject->encoding == REDIS_ENCODING_ZIPLIST &&
                ziplistLen(subject->ptr) > server.list_max_ziplist_entries)
                    listTypeConvert(subject,REDIS_ENCODING_LINKEDLIST);
338
            signalModifiedKey(c->db,c->argv[1]);
339 340 341 342 343 344 345 346
            server.dirty++;
        } else {
            /* Notify client of a failed insert */
            addReply(c,shared.cnegone);
            return;
        }
    } else {
        listTypePush(subject,val,where);
347
        signalModifiedKey(c->db,c->argv[1]);
348 349 350
        server.dirty++;
    }

351
    addReplyLongLong(c,listTypeLength(subject));
352 353 354
}

void lpushxCommand(redisClient *c) {
355
    c->argv[2] = tryObjectEncoding(c->argv[2]);
356 357 358 359
    pushxGenericCommand(c,NULL,c->argv[2],REDIS_HEAD);
}

void rpushxCommand(redisClient *c) {
360
    c->argv[2] = tryObjectEncoding(c->argv[2]);
361 362 363 364
    pushxGenericCommand(c,NULL,c->argv[2],REDIS_TAIL);
}

void linsertCommand(redisClient *c) {
365
    c->argv[4] = tryObjectEncoding(c->argv[4]);
366 367 368 369 370 371 372 373 374 375 376 377
    if (strcasecmp(c->argv[2]->ptr,"after") == 0) {
        pushxGenericCommand(c,c->argv[3],c->argv[4],REDIS_TAIL);
    } else if (strcasecmp(c->argv[2]->ptr,"before") == 0) {
        pushxGenericCommand(c,c->argv[3],c->argv[4],REDIS_HEAD);
    } else {
        addReply(c,shared.syntaxerr);
    }
}

void llenCommand(redisClient *c) {
    robj *o = lookupKeyReadOrReply(c,c->argv[1],shared.czero);
    if (o == NULL || checkType(c,o,REDIS_LIST)) return;
378
    addReplyLongLong(c,listTypeLength(o));
379 380 381 382 383
}

void lindexCommand(redisClient *c) {
    robj *o = lookupKeyReadOrReply(c,c->argv[1],shared.nullbulk);
    if (o == NULL || checkType(c,o,REDIS_LIST)) return;
384
    long index;
385 386
    robj *value = NULL;

387 388 389
    if ((getLongFromObjectOrReply(c, c->argv[2], &index, NULL) != REDIS_OK))
        return;

390 391 392 393 394 395 396 397 398 399 400 401 402 403 404 405 406 407 408 409 410 411 412 413 414 415 416 417 418 419 420 421 422
    if (o->encoding == REDIS_ENCODING_ZIPLIST) {
        unsigned char *p;
        unsigned char *vstr;
        unsigned int vlen;
        long long vlong;
        p = ziplistIndex(o->ptr,index);
        if (ziplistGet(p,&vstr,&vlen,&vlong)) {
            if (vstr) {
                value = createStringObject((char*)vstr,vlen);
            } else {
                value = createStringObjectFromLongLong(vlong);
            }
            addReplyBulk(c,value);
            decrRefCount(value);
        } else {
            addReply(c,shared.nullbulk);
        }
    } else if (o->encoding == REDIS_ENCODING_LINKEDLIST) {
        listNode *ln = listIndex(o->ptr,index);
        if (ln != NULL) {
            value = listNodeValue(ln);
            addReplyBulk(c,value);
        } else {
            addReply(c,shared.nullbulk);
        }
    } else {
        redisPanic("Unknown list encoding");
    }
}

void lsetCommand(redisClient *c) {
    robj *o = lookupKeyWriteOrReply(c,c->argv[1],shared.nokeyerr);
    if (o == NULL || checkType(c,o,REDIS_LIST)) return;
423
    long index;
424
    robj *value = (c->argv[3] = tryObjectEncoding(c->argv[3]));
425

426 427 428
    if ((getLongFromObjectOrReply(c, c->argv[2], &index, NULL) != REDIS_OK))
        return;

429 430 431 432 433 434 435 436 437 438 439 440
    listTypeTryConversion(o,value);
    if (o->encoding == REDIS_ENCODING_ZIPLIST) {
        unsigned char *p, *zl = o->ptr;
        p = ziplistIndex(zl,index);
        if (p == NULL) {
            addReply(c,shared.outofrangeerr);
        } else {
            o->ptr = ziplistDelete(o->ptr,&p);
            value = getDecodedObject(value);
            o->ptr = ziplistInsert(o->ptr,p,value->ptr,sdslen(value->ptr));
            decrRefCount(value);
            addReply(c,shared.ok);
441
            signalModifiedKey(c->db,c->argv[1]);
442 443 444 445 446 447 448 449 450 451 452
            server.dirty++;
        }
    } else if (o->encoding == REDIS_ENCODING_LINKEDLIST) {
        listNode *ln = listIndex(o->ptr,index);
        if (ln == NULL) {
            addReply(c,shared.outofrangeerr);
        } else {
            decrRefCount((robj*)listNodeValue(ln));
            listNodeValue(ln) = value;
            incrRefCount(value);
            addReply(c,shared.ok);
453
            signalModifiedKey(c->db,c->argv[1]);
454 455 456 457 458 459 460 461 462 463 464 465 466 467 468 469 470 471
            server.dirty++;
        }
    } else {
        redisPanic("Unknown list encoding");
    }
}

void popGenericCommand(redisClient *c, int where) {
    robj *o = lookupKeyWriteOrReply(c,c->argv[1],shared.nullbulk);
    if (o == NULL || checkType(c,o,REDIS_LIST)) return;

    robj *value = listTypePop(o,where);
    if (value == NULL) {
        addReply(c,shared.nullbulk);
    } else {
        addReplyBulk(c,value);
        decrRefCount(value);
        if (listTypeLength(o) == 0) dbDelete(c->db,c->argv[1]);
472
        signalModifiedKey(c->db,c->argv[1]);
473 474 475 476 477 478 479 480 481 482 483 484 485
        server.dirty++;
    }
}

void lpopCommand(redisClient *c) {
    popGenericCommand(c,REDIS_HEAD);
}

void rpopCommand(redisClient *c) {
    popGenericCommand(c,REDIS_TAIL);
}

void lrangeCommand(redisClient *c) {
486
    robj *o;
487
    long start, end, llen, rangelen;
488

489 490 491
    if ((getLongFromObjectOrReply(c, c->argv[2], &start, NULL) != REDIS_OK) ||
        (getLongFromObjectOrReply(c, c->argv[3], &end, NULL) != REDIS_OK)) return;

492 493 494 495 496 497 498 499 500
    if ((o = lookupKeyReadOrReply(c,c->argv[1],shared.emptymultibulk)) == NULL
         || checkType(c,o,REDIS_LIST)) return;
    llen = listTypeLength(o);

    /* convert negative indexes */
    if (start < 0) start = llen+start;
    if (end < 0) end = llen+end;
    if (start < 0) start = 0;

501 502
    /* Invariant: start >= 0, so this test will be true when end < 0.
     * The range is empty when start > end or start >= length. */
503 504 505 506 507 508 509 510
    if (start > end || start >= llen) {
        addReply(c,shared.emptymultibulk);
        return;
    }
    if (end >= llen) end = llen-1;
    rangelen = (end-start)+1;

    /* Return the result in form of a multi-bulk reply */
511
    addReplyMultiBulkLen(c,rangelen);
512 513 514 515 516 517 518 519 520 521 522 523 524 525 526 527
    if (o->encoding == REDIS_ENCODING_ZIPLIST) {
        unsigned char *p = ziplistIndex(o->ptr,start);
        unsigned char *vstr;
        unsigned int vlen;
        long long vlong;

        while(rangelen--) {
            ziplistGet(p,&vstr,&vlen,&vlong);
            if (vstr) {
                addReplyBulkCBuffer(c,vstr,vlen);
            } else {
                addReplyBulkLongLong(c,vlong);
            }
            p = ziplistNext(o->ptr,p);
        }
    } else if (o->encoding == REDIS_ENCODING_LINKEDLIST) {
528 529 530 531 532 533
        listNode *ln;

        /* If we are nearest to the end of the list, reach the element
         * starting from tail and going backward, as it is faster. */
        if (start > llen/2) start -= llen;
        ln = listIndex(o->ptr,start);
534 535 536 537 538 539 540

        while(rangelen--) {
            addReplyBulk(c,ln->value);
            ln = ln->next;
        }
    } else {
        redisPanic("List encoding is not LINKEDLIST nor ZIPLIST!");
541 542 543 544 545
    }
}

void ltrimCommand(redisClient *c) {
    robj *o;
546
    long start, end, llen, j, ltrim, rtrim;
547 548 549
    list *list;
    listNode *ln;

550 551 552
    if ((getLongFromObjectOrReply(c, c->argv[2], &start, NULL) != REDIS_OK) ||
        (getLongFromObjectOrReply(c, c->argv[3], &end, NULL) != REDIS_OK)) return;

553 554 555 556 557 558 559 560 561
    if ((o = lookupKeyWriteOrReply(c,c->argv[1],shared.ok)) == NULL ||
        checkType(c,o,REDIS_LIST)) return;
    llen = listTypeLength(o);

    /* convert negative indexes */
    if (start < 0) start = llen+start;
    if (end < 0) end = llen+end;
    if (start < 0) start = 0;

562 563
    /* Invariant: start >= 0, so this test will be true when end < 0.
     * The range is empty when start > end or start >= length. */
564 565 566 567 568 569 570 571 572 573 574 575 576 577 578 579 580 581 582 583 584 585 586 587 588 589 590 591
    if (start > end || start >= llen) {
        /* Out of range start or start > end result in empty list */
        ltrim = llen;
        rtrim = 0;
    } else {
        if (end >= llen) end = llen-1;
        ltrim = start;
        rtrim = llen-end-1;
    }

    /* Remove list elements to perform the trim */
    if (o->encoding == REDIS_ENCODING_ZIPLIST) {
        o->ptr = ziplistDeleteRange(o->ptr,0,ltrim);
        o->ptr = ziplistDeleteRange(o->ptr,-rtrim,rtrim);
    } else if (o->encoding == REDIS_ENCODING_LINKEDLIST) {
        list = o->ptr;
        for (j = 0; j < ltrim; j++) {
            ln = listFirst(list);
            listDelNode(list,ln);
        }
        for (j = 0; j < rtrim; j++) {
            ln = listLast(list);
            listDelNode(list,ln);
        }
    } else {
        redisPanic("Unknown list encoding");
    }
    if (listTypeLength(o) == 0) dbDelete(c->db,c->argv[1]);
592
    signalModifiedKey(c->db,c->argv[1]);
593 594 595 596 597
    server.dirty++;
    addReply(c,shared.ok);
}

void lremCommand(redisClient *c) {
598 599
    robj *subject, *obj;
    obj = c->argv[3] = tryObjectEncoding(c->argv[3]);
600
    long toremove;
601
    long removed = 0;
602 603
    listTypeEntry entry;

604 605 606
    if ((getLongFromObjectOrReply(c, c->argv[2], &toremove, NULL) != REDIS_OK))
        return;

607 608 609 610 611 612 613 614 615 616 617 618 619 620 621 622 623 624 625 626 627 628 629 630 631 632 633 634 635 636
    subject = lookupKeyWriteOrReply(c,c->argv[1],shared.czero);
    if (subject == NULL || checkType(c,subject,REDIS_LIST)) return;

    /* Make sure obj is raw when we're dealing with a ziplist */
    if (subject->encoding == REDIS_ENCODING_ZIPLIST)
        obj = getDecodedObject(obj);

    listTypeIterator *li;
    if (toremove < 0) {
        toremove = -toremove;
        li = listTypeInitIterator(subject,-1,REDIS_HEAD);
    } else {
        li = listTypeInitIterator(subject,0,REDIS_TAIL);
    }

    while (listTypeNext(li,&entry)) {
        if (listTypeEqual(&entry,obj)) {
            listTypeDelete(&entry);
            server.dirty++;
            removed++;
            if (toremove && removed == toremove) break;
        }
    }
    listTypeReleaseIterator(li);

    /* Clean up raw encoded object */
    if (subject->encoding == REDIS_ENCODING_ZIPLIST)
        decrRefCount(obj);

    if (listTypeLength(subject) == 0) dbDelete(c->db,c->argv[1]);
637
    addReplyLongLong(c,removed);
638
    if (removed) signalModifiedKey(c->db,c->argv[1]);
639 640 641 642
}

/* This is the semantic of this command:
 *  RPOPLPUSH srclist dstlist:
643 644 645 646 647 648 649
 *    IF LLEN(srclist) > 0
 *      element = RPOP srclist
 *      LPUSH dstlist element
 *      RETURN element
 *    ELSE
 *      RETURN nil
 *    END
650 651 652 653 654 655
 *  END
 *
 * The idea is to be able to get an element from a list in a reliable way
 * since the element is not just returned but pushed against another list
 * as well. This command was originally proposed by Ezra Zygmuntowicz.
 */
656

657 658 659
void rpoplpushHandlePush(redisClient *origclient, redisClient *c, robj *dstkey, robj *dstobj, robj *value) {
    robj *aux;

660
    if (!handleClientsWaitingListPush(origclient,dstkey,value)) {
661 662 663 664 665
        /* Create the list if the key does not exist */
        if (!dstobj) {
            dstobj = createZiplistObject();
            dbAdd(c->db,dstkey,dstobj);
        } else {
666
            signalModifiedKey(c->db,dstkey);
667 668
        }
        listTypePush(dstobj,value,REDIS_HEAD);
669
        /* If we are pushing as a result of LPUSH against a key
670 671 672 673
         * watched by BRPOPLPUSH, we need to rewrite the command vector
         * as an LPUSH.
         *
         * If this is called directly by RPOPLPUSH (either directly
674
         * or via a BRPOPLPUSH where the popped list exists)
675
         * we should replicate the RPOPLPUSH command itself. */
676 677 678 679 680 681 682 683 684 685 686 687
        if (c != origclient) {
            aux = createStringObject("LPUSH",5);
            rewriteClientCommandVector(origclient,3,aux,dstkey,value);
            decrRefCount(aux);
        } else {
            /* Make sure to always use RPOPLPUSH in the replication / AOF,
             * even if the original command was BRPOPLPUSH. */
            aux = createStringObject("RPOPLPUSH",9);
            rewriteClientCommandVector(origclient,3,aux,c->argv[1],c->argv[2]);
            decrRefCount(aux);
        }
        server.dirty++;
688 689 690 691 692 693
    }

    /* Always send the pushed value to the client. */
    addReplyBulk(c,value);
}

694
void rpoplpushCommand(redisClient *c) {
695 696 697 698 699 700 701 702
    robj *sobj, *value;
    if ((sobj = lookupKeyWriteOrReply(c,c->argv[1],shared.nullbulk)) == NULL ||
        checkType(c,sobj,REDIS_LIST)) return;

    if (listTypeLength(sobj) == 0) {
        addReply(c,shared.nullbulk);
    } else {
        robj *dobj = lookupKeyWrite(c->db,c->argv[2]);
703 704
        robj *touchedkey = c->argv[1];

705 706
        if (dobj && checkType(c,dobj,REDIS_LIST)) return;
        value = listTypePop(sobj,REDIS_TAIL);
707 708 709 710
        /* We saved touched key, and protect it, since rpoplpushHandlePush
         * may change the client command argument vector. */
        incrRefCount(touchedkey);
        rpoplpushHandlePush(c,c,c->argv[2],dobj,value);
711 712 713 714 715

        /* listTypePop returns an object with its refcount incremented */
        decrRefCount(value);

        /* Delete the source list when it is empty */
716 717 718
        if (listTypeLength(sobj) == 0) dbDelete(c->db,touchedkey);
        signalModifiedKey(c->db,touchedkey);
        decrRefCount(touchedkey);
719 720 721 722 723 724 725 726 727 728 729 730 731 732 733 734 735 736 737 738 739 740 741 742 743 744 745 746 747 748 749 750 751 752 753 754 755 756 757
        server.dirty++;
    }
}

/*-----------------------------------------------------------------------------
 * Blocking POP operations
 *----------------------------------------------------------------------------*/

/* Currently Redis blocking operations support is limited to list POP ops,
 * so the current implementation is not fully generic, but it is also not
 * completely specific so it will not require a rewrite to support new
 * kind of blocking operations in the future.
 *
 * Still it's important to note that list blocking operations can be already
 * used as a notification mechanism in order to implement other blocking
 * operations at application level, so there must be a very strong evidence
 * of usefulness and generality before new blocking operations are implemented.
 *
 * This is how the current blocking POP works, we use BLPOP as example:
 * - If the user calls BLPOP and the key exists and contains a non empty list
 *   then LPOP is called instead. So BLPOP is semantically the same as LPOP
 *   if there is not to block.
 * - If instead BLPOP is called and the key does not exists or the list is
 *   empty we need to block. In order to do so we remove the notification for
 *   new data to read in the client socket (so that we'll not serve new
 *   requests if the blocking request is not served). Also we put the client
 *   in a dictionary (db->blocking_keys) mapping keys to a list of clients
 *   blocking for this keys.
 * - If a PUSH operation against a key with blocked clients waiting is
 *   performed, we serve the first in the list: basically instead to push
 *   the new element inside the list we return it to the (first / oldest)
 *   blocking client, unblock the client, and remove it form the list.
 *
 * The above comment and the source code should be enough in order to understand
 * the implementation and modify / fix it later.
 */

/* Set a client in blocking mode for the specified key, with the specified
 * timeout */
758
void blockForKeys(redisClient *c, robj **keys, int numkeys, time_t timeout, robj *target) {
759 760 761 762
    dictEntry *de;
    list *l;
    int j;

763 764 765 766
    c->bpop.keys = zmalloc(sizeof(robj*)*numkeys);
    c->bpop.count = numkeys;
    c->bpop.timeout = timeout;
    c->bpop.target = target;
767 768

    if (target != NULL) {
P
Pieter Noordhuis 已提交
769
        incrRefCount(target);
770 771
    }

772 773
    for (j = 0; j < numkeys; j++) {
        /* Add the key in the client structure, to map clients -> keys */
774
        c->bpop.keys[j] = keys[j];
775 776 777 778 779 780 781 782 783 784 785
        incrRefCount(keys[j]);

        /* And in the other "side", to map keys -> clients */
        de = dictFind(c->db->blocking_keys,keys[j]);
        if (de == NULL) {
            int retval;

            /* For every key we take a list of clients blocked for it */
            l = listCreate();
            retval = dictAdd(c->db->blocking_keys,keys[j],l);
            incrRefCount(keys[j]);
786
            redisAssertWithInfo(c,keys[j],retval == DICT_OK);
787
        } else {
788
            l = dictGetVal(de);
789 790 791 792 793
        }
        listAddNodeTail(l,c);
    }
    /* Mark the client as a blocked client */
    c->flags |= REDIS_BLOCKED;
794
    server.bpop_blocked_clients++;
795 796 797 798 799 800 801 802
}

/* Unblock a client that's waiting in a blocking operation such as BLPOP */
void unblockClientWaitingData(redisClient *c) {
    dictEntry *de;
    list *l;
    int j;

803
    redisAssertWithInfo(c,NULL,c->bpop.keys != NULL);
804
    /* The client may wait for multiple keys, so unblock it for every key. */
805
    for (j = 0; j < c->bpop.count; j++) {
806
        /* Remove this client from the list of clients waiting for this key. */
807
        de = dictFind(c->db->blocking_keys,c->bpop.keys[j]);
808
        redisAssertWithInfo(c,c->bpop.keys[j],de != NULL);
809
        l = dictGetVal(de);
810 811 812
        listDelNode(l,listSearchKey(l,c));
        /* If the list is empty we need to remove it to avoid wasting memory */
        if (listLength(l) == 0)
813 814
            dictDelete(c->db->blocking_keys,c->bpop.keys[j]);
        decrRefCount(c->bpop.keys[j]);
815
    }
816

817
    /* Cleanup the client structure */
818 819
    zfree(c->bpop.keys);
    c->bpop.keys = NULL;
820
    if (c->bpop.target) decrRefCount(c->bpop.target);
821
    c->bpop.target = NULL;
822 823
    c->flags &= ~REDIS_BLOCKED;
    c->flags |= REDIS_UNBLOCKED;
824
    server.bpop_blocked_clients--;
825
    listAddNodeTail(server.unblocked_clients,c);
826 827 828 829 830 831 832 833 834 835 836 837 838 839 840
}

/* This should be called from any function PUSHing into lists.
 * 'c' is the "pushing client", 'key' is the key it is pushing data against,
 * 'ele' is the element pushed.
 *
 * If the function returns 0 there was no client waiting for a list push
 * against this key.
 *
 * If the function returns 1 there was a client waiting for a list push
 * against this key, the element was passed to this client thus it's not
 * needed to actually add it to the list and the caller should return asap. */
int handleClientsWaitingListPush(redisClient *c, robj *key, robj *ele) {
    struct dictEntry *de;
    redisClient *receiver;
841 842
    int numclients;
    list *clients;
843
    listNode *ln;
844
    robj *dstkey, *dstobj;
845 846 847

    de = dictFind(c->db->blocking_keys,key);
    if (de == NULL) return 0;
848
    clients = dictGetVal(de);
849 850 851 852 853 854 855 856 857 858 859
    numclients = listLength(clients);

    /* Try to handle the push as long as there are clients waiting for a push.
     * Note that "numclients" is used because the list of clients waiting for a
     * push on "key" is deleted by unblockClient() when empty.
     *
     * This loop will have more than 1 iteration when there is a BRPOPLPUSH
     * that cannot push the target list because it does not contain a list. If
     * this happens, it simply tries the next client waiting for a push. */
    while (numclients--) {
        ln = listFirst(clients);
860
        redisAssertWithInfo(c,key,ln != NULL);
861 862 863
        receiver = ln->value;
        dstkey = receiver->bpop.target;

864 865 866 867
        /* Protect receiver->bpop.target, that will be freed by
         * the next unblockClientWaitingData() call. */
        if (dstkey) incrRefCount(dstkey);

868 869 870 871 872 873 874 875
        /* This should remove the first element of the "clients" list. */
        unblockClientWaitingData(receiver);

        if (dstkey == NULL) {
            /* BRPOP/BLPOP */
            addReplyMultiBulkLen(receiver,2);
            addReplyBulk(receiver,key);
            addReplyBulk(receiver,ele);
876
            return 1; /* Serve just the first client as in B[RL]POP semantics */
877
        } else {
P
Pieter Noordhuis 已提交
878
            /* BRPOPLPUSH, note that receiver->db is always equal to c->db. */
879
            dstobj = lookupKeyWrite(receiver->db,dstkey);
880 881
            if (!(dstobj && checkType(receiver,dstobj,REDIS_LIST))) {
                rpoplpushHandlePush(c,receiver,dstkey,dstobj,ele);
882 883 884
                decrRefCount(dstkey);
                return 1;
            }
885
            decrRefCount(dstkey);
886
        }
D
Damian Janowski &amp; Michel Martens 已提交
887
    }
888

889
    return 0;
890 891
}

P
Pieter Noordhuis 已提交
892 893
int getTimeoutFromObjectOrReply(redisClient *c, robj *object, time_t *timeout) {
    long tval;
894

P
Pieter Noordhuis 已提交
895 896
    if (getLongFromObjectOrReply(c,object,&tval,
        "timeout is not an integer or out of range") != REDIS_OK)
897 898
        return REDIS_ERR;

P
Pieter Noordhuis 已提交
899 900
    if (tval < 0) {
        addReplyError(c,"timeout is negative");
901 902 903
        return REDIS_ERR;
    }

P
Pieter Noordhuis 已提交
904 905
    if (tval > 0) tval += time(NULL);
    *timeout = tval;
906 907

    return REDIS_OK;
908 909 910 911 912 913 914 915
}

/* Blocking RPOP/LPOP */
void blockingPopGenericCommand(redisClient *c, int where) {
    robj *o;
    time_t timeout;
    int j;

P
Pieter Noordhuis 已提交
916
    if (getTimeoutFromObjectOrReply(c,c->argv[c->argc-1],&timeout) != REDIS_OK)
917 918
        return;

919 920 921 922 923 924 925 926 927 928
    for (j = 1; j < c->argc-1; j++) {
        o = lookupKeyWrite(c->db,c->argv[j]);
        if (o != NULL) {
            if (o->type != REDIS_LIST) {
                addReply(c,shared.wrongtypeerr);
                return;
            } else {
                if (listTypeLength(o) != 0) {
                    /* If the list contains elements fall back to the usual
                     * non-blocking POP operation */
929
                    struct redisCommand *orig_cmd;
930 931 932 933 934 935 936
                    robj *argv[2], **orig_argv;
                    int orig_argc;

                    /* We need to alter the command arguments before to call
                     * popGenericCommand() as the command takes a single key. */
                    orig_argv = c->argv;
                    orig_argc = c->argc;
937
                    orig_cmd = c->cmd;
938 939 940 941 942 943 944 945 946
                    argv[1] = c->argv[j];
                    c->argv = argv;
                    c->argc = 2;

                    /* Also the return value is different, we need to output
                     * the multi bulk reply header and the key name. The
                     * "real" command will add the last element (the value)
                     * for us. If this souds like an hack to you it's just
                     * because it is... */
947
                    addReplyMultiBulkLen(c,2);
948
                    addReplyBulk(c,argv[1]);
949

950 951 952 953 954
                    popGenericCommand(c,where);

                    /* Fix the client structure with the original stuff */
                    c->argv = orig_argv;
                    c->argc = orig_argc;
955
                    c->cmd = orig_cmd;
D
Damian Janowski &amp; Michel Martens 已提交
956

957 958 959 960 961
                    return;
                }
            }
        }
    }
962

963 964 965 966 967 968 969
    /* If we are inside a MULTI/EXEC and the list is empty the only thing
     * we can do is treating it as a timeout (even with timeout 0). */
    if (c->flags & REDIS_MULTI) {
        addReply(c,shared.nullmultibulk);
        return;
    }

970
    /* If the list is empty or the key does not exists we must block */
971
    blockForKeys(c, c->argv + 1, c->argc - 2, timeout, NULL);
972 973 974 975 976 977 978 979 980
}

void blpopCommand(redisClient *c) {
    blockingPopGenericCommand(c,REDIS_HEAD);
}

void brpopCommand(redisClient *c) {
    blockingPopGenericCommand(c,REDIS_TAIL);
}
D
Damian Janowski &amp; Michel Martens 已提交
981 982

void brpoplpushCommand(redisClient *c) {
983
    time_t timeout;
D
Damian Janowski &amp; Michel Martens 已提交
984

P
Pieter Noordhuis 已提交
985
    if (getTimeoutFromObjectOrReply(c,c->argv[3],&timeout) != REDIS_OK)
986 987 988 989 990 991
        return;

    robj *key = lookupKeyWrite(c->db, c->argv[1]);

    if (key == NULL) {
        if (c->flags & REDIS_MULTI) {
992 993 994

            /* Blocking against an empty list in a multi state
             * returns immediately. */
995
            addReply(c, shared.nullbulk);
996
        } else {
997
            /* The list is empty and the client blocks. */
998 999 1000
            blockForKeys(c, c->argv + 1, 1, timeout, c->argv[2]);
        }
    } else {
1001 1002 1003 1004 1005 1006
        if (key->type != REDIS_LIST) {
            addReply(c, shared.wrongtypeerr);
        } else {

            /* The list exists and has elements, so
             * the regular rpoplpushCommand is executed. */
1007
            redisAssertWithInfo(c,key,listTypeLength(key) > 0);
1008 1009
            rpoplpushCommand(c);
        }
1010
    }
D
Damian Janowski &amp; Michel Martens 已提交
1011
}