Unverified Commit c50af0ae authored by Binbin's avatar Binbin Committed by GitHub
Browse files

Add LMPOP/BLMPOP commands. (#9373)

We want to add COUNT option for BLPOP.
But we can't do it without breaking compatibility due to the command arguments syntax.
So this commit introduce two new commands.

Syntax for the new LMPOP command:
`LMPOP numkeys [<key> ...] LEFT|RIGHT [COUNT count]`

Syntax for the new BLMPOP command:
`BLMPOP timeout numkeys [<key> ...] LEFT|RIGHT [COUNT count]`

Some background:
- LPOP takes one key, and can return multiple elements.
- BLPOP takes multiple keys, but returns one element from just one key.
- LMPOP can take multiple keys and return multiple elements from just one key.

Note that LMPOP/BLMPOP  can take multiple keys, it eventually operates on just one key.
And it will propagate as LPOP or RPOP with the COUNT option.

As a new command, it still return NIL if we can't pop any elements.
For the normal response is nested arrays in RESP2 and RESP3, like:
```
LMPOP/BLMPOP 
1) keyname
2) 1) element1
   2) element2
```
I.e. unlike BLPOP that returns a key name and one element so it uses a flat array,
and LPOP that returns multiple elements with no key name, and again uses a flat array,
this one has to return a nested array, and it does for for both RESP2 and RESP3 (like SCAN does)

Some discuss can see: #766 #8824
parent 216f168b
...@@ -65,7 +65,7 @@ ...@@ -65,7 +65,7 @@
#include "latency.h" #include "latency.h"
#include "monotonic.h" #include "monotonic.h"
int serveClientBlockedOnList(client *receiver, robj *key, robj *dstkey, redisDb *db, robj *value, int wherefrom, int whereto); void serveClientBlockedOnList(client *receiver, robj *o, robj *key, robj *dstkey, redisDb *db, int wherefrom, int whereto, int *deleted);
int getListPositionFromObjectOrReply(client *c, robj *arg, int *position); int getListPositionFromObjectOrReply(client *c, robj *arg, int *position);
/* This structure represents the blocked key information that we store /* This structure represents the blocked key information that we store
...@@ -271,6 +271,7 @@ void serveClientsBlockedOnListKey(robj *o, readyList *rl) { ...@@ -271,6 +271,7 @@ void serveClientsBlockedOnListKey(robj *o, readyList *rl) {
if (de) { if (de) {
list *clients = dictGetVal(de); list *clients = dictGetVal(de);
int numclients = listLength(clients); int numclients = listLength(clients);
int deleted = 0;
while(numclients--) { while(numclients--) {
listNode *clientnode = listFirst(clients); listNode *clientnode = listFirst(clients);
...@@ -286,41 +287,27 @@ void serveClientsBlockedOnListKey(robj *o, readyList *rl) { ...@@ -286,41 +287,27 @@ void serveClientsBlockedOnListKey(robj *o, readyList *rl) {
robj *dstkey = receiver->bpop.target; robj *dstkey = receiver->bpop.target;
int wherefrom = receiver->bpop.listpos.wherefrom; int wherefrom = receiver->bpop.listpos.wherefrom;
int whereto = receiver->bpop.listpos.whereto; int whereto = receiver->bpop.listpos.whereto;
robj *value = listTypePop(o, wherefrom);
if (value) { /* Protect receiver->bpop.target, that will be
/* Protect receiver->bpop.target, that will be * freed by the next unblockClient()
* freed by the next unblockClient() * call. */
* call. */ if (dstkey) incrRefCount(dstkey);
if (dstkey) incrRefCount(dstkey);
monotime replyTimer; monotime replyTimer;
elapsedStart(&replyTimer); elapsedStart(&replyTimer);
if (serveClientBlockedOnList(receiver, serveClientBlockedOnList(receiver, o,
rl->key,dstkey,rl->db,value, rl->key, dstkey, rl->db,
wherefrom, whereto) == C_ERR) wherefrom, whereto,
{ &deleted);
/* If we failed serving the client we need updateStatsOnUnblock(receiver, 0, elapsedUs(replyTimer));
* to also undo the POP operation. */ unblockClient(receiver);
listTypePush(o,value,wherefrom);
}
updateStatsOnUnblock(receiver, 0, elapsedUs(replyTimer));
unblockClient(receiver);
if (dstkey) decrRefCount(dstkey); if (dstkey) decrRefCount(dstkey);
decrRefCount(value);
} else {
break;
}
}
}
if (listTypeLength(o) == 0) { /* The list is empty and has been deleted. */
dbDelete(rl->db,rl->key); if (deleted) break;
notifyKeyspaceEvent(NOTIFY_GENERIC,"del",rl->key,rl->db->id); }
} }
/* We don't call signalModifiedKey() as it was already called
* when an element was pushed on the list. */
} }
/* Helper function for handleClientsBlockedOnKeys(). This function is called /* Helper function for handleClientsBlockedOnKeys(). This function is called
...@@ -627,12 +614,16 @@ void handleClientsBlockedOnKeys(void) { ...@@ -627,12 +614,16 @@ void handleClientsBlockedOnKeys(void) {
* for all the 'numkeys' keys as in the 'keys' argument. When we block for * for all the 'numkeys' keys as in the 'keys' argument. When we block for
* stream keys, we also provide an array of streamID structures: clients will * stream keys, we also provide an array of streamID structures: clients will
* be unblocked only when items with an ID greater or equal to the specified * be unblocked only when items with an ID greater or equal to the specified
* one is appended to the stream. */ * one is appended to the stream.
void blockForKeys(client *c, int btype, robj **keys, int numkeys, mstime_t timeout, robj *target, struct listPos *listpos, streamID *ids) { *
* 'count' for those commands that support the optional count argument.
* Otherwise the value is 0. */
void blockForKeys(client *c, int btype, robj **keys, int numkeys, long count, mstime_t timeout, robj *target, struct listPos *listpos, streamID *ids) {
dictEntry *de; dictEntry *de;
list *l; list *l;
int j; int j;
c->bpop.count = count;
c->bpop.timeout = timeout; c->bpop.timeout = timeout;
c->bpop.target = target; c->bpop.target = target;
......
...@@ -1703,6 +1703,16 @@ int evalGetKeys(struct redisCommand *cmd, robj **argv, int argc, getKeysResult * ...@@ -1703,6 +1703,16 @@ int evalGetKeys(struct redisCommand *cmd, robj **argv, int argc, getKeysResult *
return genericGetKeys(0, 2, 3, 1, argv, argc, result); return genericGetKeys(0, 2, 3, 1, argv, argc, result);
} }
int lmpopGetKeys(struct redisCommand *cmd, robj **argv, int argc, getKeysResult *result) {
UNUSED(cmd);
return genericGetKeys(0, 1, 2, 1, argv, argc, result);
}
int blmpopGetKeys(struct redisCommand *cmd, robj **argv, int argc, getKeysResult *result) {
UNUSED(cmd);
return genericGetKeys(0, 2, 3, 1, argv, argc, result);
}
/* Helper function to extract keys from the SORT command. /* Helper function to extract keys from the SORT command.
* *
* SORT <sort-key> ... STORE <store-key> ... * SORT <sort-key> ... STORE <store-key> ...
......
...@@ -5504,7 +5504,7 @@ RedisModuleBlockedClient *moduleBlockClient(RedisModuleCtx *ctx, RedisModuleCmdF ...@@ -5504,7 +5504,7 @@ RedisModuleBlockedClient *moduleBlockClient(RedisModuleCtx *ctx, RedisModuleCmdF
"Blocking module command called from transaction"); "Blocking module command called from transaction");
} else { } else {
if (keys) { if (keys) {
blockForKeys(c,BLOCKED_MODULE,keys,numkeys,timeout,NULL,NULL,NULL); blockForKeys(c,BLOCKED_MODULE,keys,numkeys,0,timeout,NULL,NULL,NULL);
} else { } else {
blockClient(c,BLOCKED_MODULE); blockClient(c,BLOCKED_MODULE);
} }
......
...@@ -316,6 +316,10 @@ struct redisCommand redisCommandTable[] = { ...@@ -316,6 +316,10 @@ struct redisCommand redisCommandTable[] = {
"write fast @list", "write fast @list",
0,NULL,1,1,1,0,0,0}, 0,NULL,1,1,1,0,0,0},
{"lmpop",lmpopCommand,-4,
"write @list",
0,lmpopGetKeys,0,0,0,0,0,0},
{"brpop",brpopCommand,-3, {"brpop",brpopCommand,-3,
"write no-script @list @blocking", "write no-script @list @blocking",
0,NULL,1,-2,1,0,0,0}, 0,NULL,1,-2,1,0,0,0},
...@@ -332,6 +336,10 @@ struct redisCommand redisCommandTable[] = { ...@@ -332,6 +336,10 @@ struct redisCommand redisCommandTable[] = {
"write no-script @list @blocking", "write no-script @list @blocking",
0,NULL,1,-2,1,0,0,0}, 0,NULL,1,-2,1,0,0,0},
{"blmpop",blmpopCommand,-5,
"write @list @blocking",
0,blmpopGetKeys,0,0,0,0,0,0},
{"llen",llenCommand,2, {"llen",llenCommand,2,
"read-only fast @list", "read-only fast @list",
0,NULL,1,1,1,0,0,0}, 0,NULL,1,1,1,0,0,0},
......
...@@ -793,6 +793,7 @@ typedef struct multiState { ...@@ -793,6 +793,7 @@ typedef struct multiState {
* The fields used depend on client->btype. */ * The fields used depend on client->btype. */
typedef struct blockingState { typedef struct blockingState {
/* Generic fields. */ /* Generic fields. */
long count; /* Elements to pop if count was specified (BLMPOP), 0 otherwise. */
mstime_t timeout; /* Blocking operation timeout. If UNIX current time mstime_t timeout; /* Blocking operation timeout. If UNIX current time
* is > timeout then the operation timed out. */ * is > timeout then the operation timed out. */
...@@ -1910,6 +1911,7 @@ void addReplyPushLen(client *c, long length); ...@@ -1910,6 +1911,7 @@ void addReplyPushLen(client *c, long length);
void addReplyHelp(client *c, const char **help); void addReplyHelp(client *c, const char **help);
void addReplySubcommandSyntaxError(client *c); void addReplySubcommandSyntaxError(client *c);
void addReplyLoadedModules(client *c); void addReplyLoadedModules(client *c);
void addListRangeReply(client *c, robj *o, long start, long end, int reverse);
void copyClientOutputBuffer(client *dst, client *src); void copyClientOutputBuffer(client *dst, client *src);
size_t sdsZmallocSize(sds s); size_t sdsZmallocSize(sds s);
size_t getStringObjectSdsUsedMemory(robj *o); size_t getStringObjectSdsUsedMemory(robj *o);
...@@ -1991,9 +1993,10 @@ int listTypeEqual(listTypeEntry *entry, robj *o); ...@@ -1991,9 +1993,10 @@ int listTypeEqual(listTypeEntry *entry, robj *o);
void listTypeDelete(listTypeIterator *iter, listTypeEntry *entry); void listTypeDelete(listTypeIterator *iter, listTypeEntry *entry);
void listTypeConvert(robj *subject, int enc); void listTypeConvert(robj *subject, int enc);
robj *listTypeDup(robj *o); robj *listTypeDup(robj *o);
int listTypeDelRange(robj *o, long start, long stop);
void unblockClientWaitingData(client *c); void unblockClientWaitingData(client *c);
void popGenericCommand(client *c, int where); void popGenericCommand(client *c, int where);
void listElementsRemoved(client *c, robj *key, int where, robj *o, long count); void listElementsRemoved(client *c, robj *key, int where, robj *o, long count, int *deleted);
/* MULTI/EXEC/WATCH... */ /* MULTI/EXEC/WATCH... */
void unwatchAllKeys(client *c); void unwatchAllKeys(client *c);
...@@ -2451,6 +2454,8 @@ int georadiusGetKeys(struct redisCommand *cmd, robj **argv, int argc, getKeysRes ...@@ -2451,6 +2454,8 @@ int georadiusGetKeys(struct redisCommand *cmd, robj **argv, int argc, getKeysRes
int xreadGetKeys(struct redisCommand *cmd, robj **argv, int argc, getKeysResult *result); int xreadGetKeys(struct redisCommand *cmd, robj **argv, int argc, getKeysResult *result);
int memoryGetKeys(struct redisCommand *cmd, robj **argv, int argc, getKeysResult *result); int memoryGetKeys(struct redisCommand *cmd, robj **argv, int argc, getKeysResult *result);
int lcsGetKeys(struct redisCommand *cmd, robj **argv, int argc, getKeysResult *result); int lcsGetKeys(struct redisCommand *cmd, robj **argv, int argc, getKeysResult *result);
int lmpopGetKeys(struct redisCommand *cmd, robj **argv, int argc, getKeysResult *result);
int blmpopGetKeys(struct redisCommand *cmd, robj **argv, int argc, getKeysResult *result);
unsigned short crc16(const char *buf, int len); unsigned short crc16(const char *buf, int len);
...@@ -2487,7 +2492,7 @@ int getTimeoutFromObjectOrReply(client *c, robj *object, mstime_t *timeout, int ...@@ -2487,7 +2492,7 @@ int getTimeoutFromObjectOrReply(client *c, robj *object, mstime_t *timeout, int
void disconnectAllBlockedClients(void); void disconnectAllBlockedClients(void);
void handleClientsBlockedOnKeys(void); void handleClientsBlockedOnKeys(void);
void signalKeyAsReady(redisDb *db, robj *key, int type); void signalKeyAsReady(redisDb *db, robj *key, int type);
void blockForKeys(client *c, int btype, robj **keys, int numkeys, mstime_t timeout, robj *target, struct listPos *listpos, streamID *ids); void blockForKeys(client *c, int btype, robj **keys, int numkeys, long count, mstime_t timeout, robj *target, struct listPos *listpos, streamID *ids);
void updateStatsOnUnblock(client *c, long blocked_us, long reply_us); void updateStatsOnUnblock(client *c, long blocked_us, long reply_us);
/* timeout.c -- Blocked clients timeout and connections timeout. */ /* timeout.c -- Blocked clients timeout and connections timeout. */
...@@ -2576,6 +2581,7 @@ void rpushxCommand(client *c); ...@@ -2576,6 +2581,7 @@ void rpushxCommand(client *c);
void linsertCommand(client *c); void linsertCommand(client *c);
void lpopCommand(client *c); void lpopCommand(client *c);
void rpopCommand(client *c); void rpopCommand(client *c);
void lmpopCommand(client *c);
void llenCommand(client *c); void llenCommand(client *c);
void lindexCommand(client *c); void lindexCommand(client *c);
void lrangeCommand(client *c); void lrangeCommand(client *c);
...@@ -2652,6 +2658,7 @@ void execCommand(client *c); ...@@ -2652,6 +2658,7 @@ void execCommand(client *c);
void discardCommand(client *c); void discardCommand(client *c);
void blpopCommand(client *c); void blpopCommand(client *c);
void brpopCommand(client *c); void brpopCommand(client *c);
void blmpopCommand(client *c);
void brpoplpushCommand(client *c); void brpoplpushCommand(client *c);
void blmoveCommand(client *c); void blmoveCommand(client *c);
void appendCommand(client *c); void appendCommand(client *c);
......
...@@ -215,6 +215,15 @@ robj *listTypeDup(robj *o) { ...@@ -215,6 +215,15 @@ robj *listTypeDup(robj *o) {
return lobj; return lobj;
} }
/* Delete a range of elements from the list. */
int listTypeDelRange(robj *subject, long start, long count) {
if (subject->encoding == OBJ_ENCODING_QUICKLIST) {
return quicklistDelRange(subject->ptr, start, count);
} else {
serverPanic("Unknown list encoding");
}
}
/*----------------------------------------------------------------------------- /*-----------------------------------------------------------------------------
* List Commands * List Commands
*----------------------------------------------------------------------------*/ *----------------------------------------------------------------------------*/
...@@ -374,6 +383,35 @@ void lsetCommand(client *c) { ...@@ -374,6 +383,35 @@ void lsetCommand(client *c) {
} }
} }
/* A helper function like addListRangeReply, more details see below.
* The difference is that here we are returning nested arrays, like:
* 1) keyname
* 2) 1) element1
* 2) element2
*
* And also actually pop out from the list by calling listElementsRemoved.
* We maintain the server.dirty and notifications there.
*
* 'deleted' is an optional output argument to get an indication
* if the key got deleted by this function. */
void listPopRangeAndReplyWithKey(client *c, robj *o, robj *key, int where, long count, int *deleted) {
long llen = listTypeLength(o);
long rangelen = (count > llen) ? llen : count;
long rangestart = (where == LIST_HEAD) ? 0 : -rangelen;
long rangeend = (where == LIST_HEAD) ? rangelen - 1 : -1;
int reverse = (where == LIST_HEAD) ? 0 : 1;
/* We return key-name just once, and an array of elements */
addReplyArrayLen(c, 2);
addReplyBulk(c, key);
addListRangeReply(c, o, rangestart, rangeend, reverse);
/* Pop these elements. */
listTypeDelRange(o, rangestart, rangelen);
/* Maintain the notifications and dirty. */
listElementsRemoved(c, key, where, o, rangelen, deleted);
}
/* A helper for replying with a list's range between the inclusive start and end /* A helper for replying with a list's range between the inclusive start and end
* indexes as multi-bulk, with support for negative indexes. Note that start * indexes as multi-bulk, with support for negative indexes. Note that start
* must be less than end or an empty array is returned. When the reverse * must be less than end or an empty array is returned. When the reverse
...@@ -419,14 +457,21 @@ void addListRangeReply(client *c, robj *o, long start, long end, int reverse) { ...@@ -419,14 +457,21 @@ void addListRangeReply(client *c, robj *o, long start, long end, int reverse) {
} }
} }
/* A housekeeping helper for list elements popping tasks. */ /* A housekeeping helper for list elements popping tasks.
void listElementsRemoved(client *c, robj *key, int where, robj *o, long count) { *
* 'deleted' is an optional output argument to get an indication
* if the key got deleted by this function. */
void listElementsRemoved(client *c, robj *key, int where, robj *o, long count, int *deleted) {
char *event = (where == LIST_HEAD) ? "lpop" : "rpop"; char *event = (where == LIST_HEAD) ? "lpop" : "rpop";
notifyKeyspaceEvent(NOTIFY_LIST, event, key, c->db->id); notifyKeyspaceEvent(NOTIFY_LIST, event, key, c->db->id);
if (listTypeLength(o) == 0) { if (listTypeLength(o) == 0) {
notifyKeyspaceEvent(NOTIFY_GENERIC, "del", key, c->db->id); if (deleted) *deleted = 1;
dbDelete(c->db, key); dbDelete(c->db, key);
notifyKeyspaceEvent(NOTIFY_GENERIC, "del", key, c->db->id);
} else {
if (deleted) *deleted = 0;
} }
signalModifiedKey(c, c->db, key); signalModifiedKey(c, c->db, key);
server.dirty += count; server.dirty += count;
...@@ -466,7 +511,7 @@ void popGenericCommand(client *c, int where) { ...@@ -466,7 +511,7 @@ void popGenericCommand(client *c, int where) {
serverAssert(value != NULL); serverAssert(value != NULL);
addReplyBulk(c,value); addReplyBulk(c,value);
decrRefCount(value); decrRefCount(value);
listElementsRemoved(c,c->argv[1],where,o,1); listElementsRemoved(c,c->argv[1],where,o,1,NULL);
} else { } else {
/* Pop a range of elements. An addition to the original POP command, /* Pop a range of elements. An addition to the original POP command,
* which replies with a multi-bulk. */ * which replies with a multi-bulk. */
...@@ -477,9 +522,50 @@ void popGenericCommand(client *c, int where) { ...@@ -477,9 +522,50 @@ void popGenericCommand(client *c, int where) {
int reverse = (where == LIST_HEAD) ? 0 : 1; int reverse = (where == LIST_HEAD) ? 0 : 1;
addListRangeReply(c,o,rangestart,rangeend,reverse); addListRangeReply(c,o,rangestart,rangeend,reverse);
quicklistDelRange(o->ptr,rangestart,rangelen); listTypeDelRange(o,rangestart,rangelen);
listElementsRemoved(c,c->argv[1],where,o,rangelen); listElementsRemoved(c,c->argv[1],where,o,rangelen,NULL);
}
}
/* Like popGenericCommand but work with multiple keys.
* Take multiple keys and return multiple elements from just one key.
*
* 'numkeys' the number of keys.
* 'count' is the number of elements requested to pop.
*
* Always reply with array. */
void mpopGenericCommand(client *c, robj **keys, int numkeys, int where, long count) {
int j;
robj *o;
robj *key;
for (j = 0; j < numkeys; j++) {
key = keys[j];
o = lookupKeyWrite(c->db, key);
/* Non-existing key, move to next key. */
if (o == NULL) continue;
if (checkType(c, o, OBJ_LIST)) return;
long llen = listTypeLength(o);
/* Empty list, move to next key. */
if (llen == 0) continue;
/* Pop a range of elements in a nested arrays way. */
listPopRangeAndReplyWithKey(c, o, key, where, count, NULL);
/* Replicate it as [LR]POP COUNT. */
robj *count_obj = createStringObjectFromLongLong((count > llen) ? llen : count);
rewriteClientCommandVector(c, 3,
(where == LIST_HEAD) ? shared.lpop : shared.rpop,
key, count_obj);
decrRefCount(count_obj);
return;
} }
/* Look like we are not able to pop up any elements. */
addReplyNullArray(c);
} }
/* LPOP <key> [count] */ /* LPOP <key> [count] */
...@@ -829,10 +915,11 @@ void rpoplpushCommand(client *c) { ...@@ -829,10 +915,11 @@ void rpoplpushCommand(client *c) {
* is to serve a specific client (receiver) that is blocked on 'key' * is to serve a specific client (receiver) that is blocked on 'key'
* in the context of the specified 'db', doing the following: * in the context of the specified 'db', doing the following:
* *
* 1) Provide the client with the 'value' element. * 1) Provide the client with the 'value' element or a range of elements.
* We will do the pop in here and caller does not need to bother the return.
* 2) If the dstkey is not NULL (we are serving a BLMOVE) also push the * 2) If the dstkey is not NULL (we are serving a BLMOVE) also push the
* 'value' element on the destination list (the "push" side of the command). * 'value' element on the destination list (the "push" side of the command).
* 3) Propagate the resulting BRPOP, BLPOP and additional xPUSH if any into * 3) Propagate the resulting BRPOP, BLPOP, BLMPOP and additional xPUSH if any into
* the AOF and replication channel. * the AOF and replication channel.
* *
* The argument 'wherefrom' is LIST_TAIL or LIST_HEAD, and indicates if the * The argument 'wherefrom' is LIST_TAIL or LIST_HEAD, and indicates if the
...@@ -843,25 +930,45 @@ void rpoplpushCommand(client *c) { ...@@ -843,25 +930,45 @@ void rpoplpushCommand(client *c) {
* 'value' element is to be pushed to the head or tail so that we can * 'value' element is to be pushed to the head or tail so that we can
* propagate the command properly. * propagate the command properly.
* *
* The function returns C_OK if we are able to serve the client, otherwise * 'deleted' is an optional output argument to get an indication
* C_ERR is returned to signal the caller that the list POP operation * if the key got deleted by this function. */
* should be undone as the client was not served: This only happens for void serveClientBlockedOnList(client *receiver, robj *o, robj *key, robj *dstkey, redisDb *db, int wherefrom, int whereto, int *deleted)
* BLMOVE that fails to push the value to the destination key as it is
* of the wrong type. */
int serveClientBlockedOnList(client *receiver, robj *key, robj *dstkey, redisDb *db, robj *value, int wherefrom, int whereto)
{ {
robj *argv[5]; robj *argv[5];
robj *value = NULL;
if (deleted) *deleted = 0;
if (dstkey == NULL) { if (dstkey == NULL) {
/* Propagate the [LR]POP operation. */ /* Propagate the [LR]POP operation. */
struct redisCommand *cmd = (wherefrom == LIST_HEAD) ?
server.lpopCommand : server.rpopCommand;
argv[0] = (wherefrom == LIST_HEAD) ? shared.lpop : argv[0] = (wherefrom == LIST_HEAD) ? shared.lpop :
shared.rpop; shared.rpop;
argv[1] = key; argv[1] = key;
propagate((wherefrom == LIST_HEAD) ?
server.lpopCommand : server.rpopCommand, if (receiver->lastcmd->proc == blmpopCommand) {
db->id,argv,2,PROPAGATE_AOF|PROPAGATE_REPL); /* Propagate the [LR]POP COUNT operation. */
long count = receiver->bpop.count;
serverAssert(count > 0);
long llen = listTypeLength(o);
serverAssert(llen > 0);
argv[2] = createStringObjectFromLongLong((count > llen) ? llen : count);
propagate(cmd, db->id, argv, 3, PROPAGATE_AOF|PROPAGATE_REPL);
decrRefCount(argv[2]);
/* Pop a range of elements in a nested arrays way. */
listPopRangeAndReplyWithKey(receiver, o, key, wherefrom, count, deleted);
return;
}
propagate(cmd, db->id, argv, 2, PROPAGATE_AOF|PROPAGATE_REPL);
/* BRPOP/BLPOP */ /* BRPOP/BLPOP */
value = listTypePop(o, wherefrom);
serverAssert(value != NULL);
addReplyArrayLen(receiver,2); addReplyArrayLen(receiver,2);
addReplyBulk(receiver,key); addReplyBulk(receiver,key);
addReplyBulk(receiver,value); addReplyBulk(receiver,value);
...@@ -876,6 +983,9 @@ int serveClientBlockedOnList(client *receiver, robj *key, robj *dstkey, redisDb ...@@ -876,6 +983,9 @@ int serveClientBlockedOnList(client *receiver, robj *key, robj *dstkey, redisDb
if (!(dstobj && if (!(dstobj &&
checkType(receiver,dstobj,OBJ_LIST))) checkType(receiver,dstobj,OBJ_LIST)))
{ {
value = listTypePop(o, wherefrom);
serverAssert(value != NULL);
lmoveHandlePush(receiver,dstkey,dstobj,value,whereto); lmoveHandlePush(receiver,dstkey,dstobj,value,whereto);
/* Propagate the LMOVE/RPOPLPUSH operation. */ /* Propagate the LMOVE/RPOPLPUSH operation. */
int isbrpoplpush = (receiver->lastcmd->proc == brpoplpushCommand); int isbrpoplpush = (receiver->lastcmd->proc == brpoplpushCommand);
...@@ -892,49 +1002,82 @@ int serveClientBlockedOnList(client *receiver, robj *key, robj *dstkey, redisDb ...@@ -892,49 +1002,82 @@ int serveClientBlockedOnList(client *receiver, robj *key, robj *dstkey, redisDb
/* Notify event ("lpush" or "rpush" was notified by lmoveHandlePush). */ /* Notify event ("lpush" or "rpush" was notified by lmoveHandlePush). */
notifyKeyspaceEvent(NOTIFY_LIST,wherefrom == LIST_TAIL ? "rpop" : "lpop", notifyKeyspaceEvent(NOTIFY_LIST,wherefrom == LIST_TAIL ? "rpop" : "lpop",
key,receiver->db->id); key,receiver->db->id);
} else {
/* BLMOVE failed because of wrong
* destination type. */
return C_ERR;
} }
} }
return C_OK;
if (value) decrRefCount(value);
if (listTypeLength(o) == 0) {
if (deleted) *deleted = 1;
dbDelete(receiver->db, key);
notifyKeyspaceEvent(NOTIFY_GENERIC, "del", key, receiver->db->id);
}
/* We don't call signalModifiedKey() as it was already called
* when an element was pushed on the list. */
} }
/* Blocking RPOP/LPOP */ /* Blocking RPOP/LPOP/LMPOP
void blockingPopGenericCommand(client *c, int where) { *
* 'numkeys' is the number of keys.
* 'timeout_idx' parameter position of block timeout.
* 'where' LIST_HEAD for LEFT, LIST_TAIL for RIGHT.
* 'count' is the number of elements requested to pop, or 0 for plain single pop.
*
* When count is 0, a reply of a single bulk-string will be used.
* When count > 0, an array reply will be used. */
void blockingPopGenericCommand(client *c, robj **keys, int numkeys, int where, int timeout_idx, long count) {
robj *o; robj *o;
robj *key;
mstime_t timeout; mstime_t timeout;
int j; int j;
if (getTimeoutFromObjectOrReply(c,c->argv[c->argc-1],&timeout,UNIT_SECONDS) if (getTimeoutFromObjectOrReply(c,c->argv[timeout_idx],&timeout,UNIT_SECONDS)
!= C_OK) return; != C_OK) return;
for (j = 1; j < c->argc-1; j++) { /* Traverse all input keys, we take action only based on one key. */
o = lookupKeyWrite(c->db,c->argv[j]); for (j = 0; j < numkeys; j++) {
if (o != NULL) { key = keys[j];
if (checkType(c,o,OBJ_LIST)) { o = lookupKeyWrite(c->db, key);
return;
} else { /* Non-existing key, move to next key. */
if (listTypeLength(o) != 0) { if (o == NULL) continue;
/* Non empty list, this is like a normal [LR]POP. */
robj *value = listTypePop(o,where); if (checkType(c, o, OBJ_LIST)) return;
serverAssert(value != NULL);
long llen = listTypeLength(o);
addReplyArrayLen(c,2); /* Empty list, move to next key. */
addReplyBulk(c,c->argv[j]); if (llen == 0) continue;
addReplyBulk(c,value);
decrRefCount(value); if (count != 0) {
listElementsRemoved(c,c->argv[j],where,o,1); /* BLMPOP, non empty list, like a normal [LR]POP with count option.
* The difference here we pop a range of elements in a nested arrays way. */
/* Replicate it as an [LR]POP instead of B[LR]POP. */ listPopRangeAndReplyWithKey(c, o, key, where, count, NULL);
rewriteClientCommandVector(c,2,
(where == LIST_HEAD) ? shared.lpop : shared.rpop, /* Replicate it as [LR]POP COUNT. */
c->argv[j]); robj *count_obj = createStringObjectFromLongLong((count > llen) ? llen : count);
return; rewriteClientCommandVector(c, 3,
} (where == LIST_HEAD) ? shared.lpop : shared.rpop,
} key, count_obj);
decrRefCount(count_obj);
return;
} }
/* Non empty list, this is like a normal [LR]POP. */
robj *value = listTypePop(o,where);
serverAssert(value != NULL);
addReplyArrayLen(c,2);
addReplyBulk(c,key);
addReplyBulk(c,value);
decrRefCount(value);
listElementsRemoved(c,key,where,o,1,NULL);
/* Replicate it as an [LR]POP instead of B[LR]POP. */
rewriteClientCommandVector(c,2,
(where == LIST_HEAD) ? shared.lpop : shared.rpop,
key);
return;
} }
/* If we are not allowed to block the client, the only thing /* If we are not allowed to block the client, the only thing
...@@ -946,17 +1089,17 @@ void blockingPopGenericCommand(client *c, int where) { ...@@ -946,17 +1089,17 @@ void blockingPopGenericCommand(client *c, int where) {
/* If the keys do not exist we must block */ /* If the keys do not exist we must block */
struct listPos pos = {where}; struct listPos pos = {where};
blockForKeys(c,BLOCKED_LIST,c->argv + 1,c->argc - 2,timeout,NULL,&pos,NULL); blockForKeys(c,BLOCKED_LIST,keys,numkeys,count,timeout,NULL,&pos,NULL);
} }
/* BLPOP <key> [<key> ...] <timeout> */ /* BLPOP <key> [<key> ...] <timeout> */
void blpopCommand(client *c) { void blpopCommand(client *c) {
blockingPopGenericCommand(c,LIST_HEAD); blockingPopGenericCommand(c,c->argv+1,c->argc-2,LIST_HEAD,c->argc-1,0);
} }
/* BRPOP <key> [<key> ...] <timeout> */ /* BRPOP <key> [<key> ...] <timeout> */
void brpopCommand(client *c) { void brpopCommand(client *c) {
blockingPopGenericCommand(c,LIST_TAIL); blockingPopGenericCommand(c,c->argv+1,c->argc-2,LIST_TAIL,c->argc-1,0);
} }
void blmoveGenericCommand(client *c, int wherefrom, int whereto, mstime_t timeout) { void blmoveGenericCommand(client *c, int wherefrom, int whereto, mstime_t timeout) {
...@@ -971,7 +1114,7 @@ void blmoveGenericCommand(client *c, int wherefrom, int whereto, mstime_t timeou ...@@ -971,7 +1114,7 @@ void blmoveGenericCommand(client *c, int wherefrom, int whereto, mstime_t timeou
} else { } else {
/* The list is empty and the client blocks. */ /* The list is empty and the client blocks. */
struct listPos pos = {wherefrom, whereto}; struct listPos pos = {wherefrom, whereto};
blockForKeys(c,BLOCKED_LIST,c->argv + 1,1,timeout,c->argv[2],&pos,NULL); blockForKeys(c,BLOCKED_LIST,c->argv + 1,1,0,timeout,c->argv[2],&pos,NULL);
} }
} else { } else {
/* The list exists and has elements, so /* The list exists and has elements, so
...@@ -1001,3 +1144,62 @@ void brpoplpushCommand(client *c) { ...@@ -1001,3 +1144,62 @@ void brpoplpushCommand(client *c) {
!= C_OK) return; != C_OK) return;
blmoveGenericCommand(c, LIST_TAIL, LIST_HEAD, timeout); blmoveGenericCommand(c, LIST_TAIL, LIST_HEAD, timeout);
} }
/* LMPOP/BLMPOP
*
* 'numkeys_idx' parameter position of key number.
* 'is_block' this indicates whether it is a blocking variant. */
void lmpopGenericCommand(client *c, int numkeys_idx, int is_block) {
long j;
long numkeys = 0; /* Number of keys. */
int where = 0; /* HEAD for LEFT, TAIL for RIGHT. */
long count = 1; /* Reply will consist of up to count elements, depending on the list's length. */
/* Parse the numkeys. */
if (getRangeLongFromObjectOrReply(c, c->argv[numkeys_idx], 1, LONG_MAX,
&numkeys, "numkeys should be greater than 0") != C_OK)
return;
/* Parse the where. where_idx: the index of where in the c->argv. */
long where_idx = numkeys_idx + numkeys + 1;
if (where_idx >= c->argc) {
addReplyErrorObject(c, shared.syntaxerr);
return;
}
if (getListPositionFromObjectOrReply(c, c->argv[where_idx], &where) != C_OK)
return;
/* Parse the optional arguments. */
for (j = where_idx + 1; j < c->argc; j++) {
char *opt = c->argv[j]->ptr;
int moreargs = (c->argc - 1) - j;
if (!strcasecmp(opt, "COUNT") && moreargs) {
j++;
if (getRangeLongFromObjectOrReply(c, c->argv[j], 1, LONG_MAX,
&count,"count should be greater than 0") != C_OK)
return;
} else {
addReplyErrorObject(c, shared.syntaxerr);
return;
}
}
if (is_block) {
/* BLOCK. We will handle CLIENT_DENY_BLOCKING flag in blockingPopGenericCommand. */
blockingPopGenericCommand(c, c->argv+numkeys_idx+1, numkeys, where, 1, count);
} else {
/* NON-BLOCK */
mpopGenericCommand(c, c->argv+numkeys_idx+1, numkeys, where, count);
}
}
/* LMPOP numkeys [<key> ...] LEFT|RIGHT [COUNT count] */
void lmpopCommand(client *c) {
lmpopGenericCommand(c, 1, 0);
}
/* BLMPOP timeout numkeys [<key> ...] LEFT|RIGHT [COUNT count] */
void blmpopCommand(client *c) {
lmpopGenericCommand(c, 2, 1);
}
...@@ -2152,7 +2152,7 @@ void xreadCommand(client *c) { ...@@ -2152,7 +2152,7 @@ void xreadCommand(client *c) {
goto cleanup; goto cleanup;
} }
blockForKeys(c, BLOCKED_STREAM, c->argv+streams_arg, streams_count, blockForKeys(c, BLOCKED_STREAM, c->argv+streams_arg, streams_count,
timeout, NULL, NULL, ids); 0, timeout, NULL, NULL, ids);
/* If no COUNT is given and we block, set a relatively small count: /* If no COUNT is given and we block, set a relatively small count:
* in case the ID provided is too low, we do not want the server to * in case the ID provided is too low, we do not want the server to
* block just to serve this client a huge stream of messages. */ * block just to serve this client a huge stream of messages. */
......
...@@ -3967,7 +3967,7 @@ void blockingGenericZpopCommand(client *c, int where) { ...@@ -3967,7 +3967,7 @@ void blockingGenericZpopCommand(client *c, int where) {
} }
/* If the keys do not exist we must block */ /* If the keys do not exist we must block */
blockForKeys(c,BLOCKED_ZSET,c->argv + 1,c->argc - 2,timeout,NULL,NULL,NULL); blockForKeys(c,BLOCKED_ZSET,c->argv + 1,c->argc - 2,0,timeout,NULL,NULL,NULL);
} }
// BZPOPMIN key [key ...] timeout // BZPOPMIN key [key ...] timeout
......
...@@ -320,4 +320,52 @@ tags {"aof external:skip"} { ...@@ -320,4 +320,52 @@ tags {"aof external:skip"} {
} }
} }
} }
# Test that LMPOP/BLMPOP work fine with AOF.
create_aof {
append_to_aof [formatCommand lpush mylist a b c]
append_to_aof [formatCommand rpush mylist2 1 2 3]
append_to_aof [formatCommand lpush mylist3 a b c d e]
}
start_server_aof [list dir $server_path aof-load-truncated no] {
test "AOF+LMPOP/BLMPOP: pop elements from the list" {
set client [redis [dict get $srv host] [dict get $srv port] 0 $::tls]
set client2 [redis [dict get $srv host] [dict get $srv port] 1 $::tls]
wait_done_loading $client
wait_done_loading $client2
# Pop all elements from mylist, should be blmpop delete mylist.
$client lmpop 1 mylist left count 1
$client blmpop 0 1 mylist left count 10
# Pop all elements from mylist2, should be lmpop delete mylist2.
$client blmpop 0 2 mylist mylist2 right count 10
$client lmpop 2 mylist mylist2 right count 2
# Blocking path, be blocked and then released.
$client2 blmpop 0 2 mylist mylist2 left count 2
after 100
$client lpush mylist2 a b c
# Pop up the last element in mylist2
$client blmpop 0 3 mylist mylist2 mylist3 left count 1
# Leave two elements in mylist3.
$client blmpop 0 3 mylist mylist2 mylist3 right count 3
}
}
start_server_aof [list dir $server_path aof-load-truncated no] {
test "AOF+LMPOP/BLMPOP: after pop elements from the list" {
set client [redis [dict get $srv host] [dict get $srv port] 0 $::tls]
wait_done_loading $client
# mylist and mylist2 no longer exist.
assert_equal 0 [$client exists mylist mylist2]
# Length of mylist3 is two.
assert_equal 2 [$client llen mylist3]
}
}
} }
This diff is collapsed.
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment