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