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

Add ZMPOP/BZMPOP commands. (#9484)

This is similar to the recent addition of LMPOP/BLMPOP (#9373), but zset.

Syntax for the new ZMPOP command:
`ZMPOP numkeys [<key> ...] MIN|MAX [COUNT count]`

Syntax for the new BZMPOP command:
`BZMPOP timeout numkeys [<key> ...] MIN|MAX [COUNT count]`

Some background:
- ZPOPMIN/ZPOPMAX take only one key, and can return multiple elements.
- BZPOPMIN/BZPOPMAX take multiple keys, but return only one element from just one key.
- ZMPOP/BZMPOP can take multiple keys, and can return multiple elements from just one key.

Note that ZMPOP/BZMPOP can take multiple keys, it eventually operates on just on key.
And it will propagate as ZPOPMIN or ZPOPMAX with the COUNT option.

As new commands, if we can not pop any elements, the response like:
- ZMPOP: Return a NIL in both RESP2 and RESP3, unlike ZPOPMIN/ZPOPMAX return emptyarray.
- BZMPOP: Return a NIL in both RESP2 and RESP3 when timeout is reached, like BZPOPMIN/BZPOPMAX.

For the normal response is nested arrays in RESP2 and RESP3:
```
ZMPOP/BZMPOP
1) keyname
2) 1) 1) member1
      2) score1
   2) 1) member2
      2) score2

In RESP2:
1) "myzset"
2) 1) 1) "three"
      2) "3"
   2) 1) "two"
      2) "2"

In RESP3:
1) "myzset"
2) 1) 1) "three"
      2) (double) 3
   2) 1) "two"
      2) (double) 2
```
parent 59c9716f
...@@ -285,8 +285,8 @@ void serveClientsBlockedOnListKey(robj *o, readyList *rl) { ...@@ -285,8 +285,8 @@ 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.blockpos.wherefrom;
int whereto = receiver->bpop.listpos.whereto; int whereto = receiver->bpop.blockpos.whereto;
/* Protect receiver->bpop.target, that will be /* Protect receiver->bpop.target, that will be
* freed by the next unblockClient() * freed by the next unblockClient()
...@@ -320,9 +320,9 @@ void serveClientsBlockedOnSortedSetKey(robj *o, readyList *rl) { ...@@ -320,9 +320,9 @@ void serveClientsBlockedOnSortedSetKey(robj *o, readyList *rl) {
if (de) { if (de) {
list *clients = dictGetVal(de); list *clients = dictGetVal(de);
int numclients = listLength(clients); int numclients = listLength(clients);
unsigned long zcard = zsetLength(o); int deleted = 0;
while(numclients-- && zcard) { while (numclients--) {
listNode *clientnode = listFirst(clients); listNode *clientnode = listFirst(clients);
client *receiver = clientnode->value; client *receiver = clientnode->value;
...@@ -333,23 +333,38 @@ void serveClientsBlockedOnSortedSetKey(robj *o, readyList *rl) { ...@@ -333,23 +333,38 @@ void serveClientsBlockedOnSortedSetKey(robj *o, readyList *rl) {
continue; continue;
} }
int where = (receiver->lastcmd && long llen = zsetLength(o);
receiver->lastcmd->proc == bzpopminCommand) long count = receiver->bpop.count;
? ZSET_MIN : ZSET_MAX; int where = receiver->bpop.blockpos.wherefrom;
int use_nested_array = (receiver->lastcmd &&
receiver->lastcmd->proc == bzmpopCommand)
? 1 : 0;
int reply_nil_when_empty = use_nested_array;
monotime replyTimer; monotime replyTimer;
elapsedStart(&replyTimer); elapsedStart(&replyTimer);
genericZpopCommand(receiver,&rl->key,1,where,1,NULL); genericZpopCommand(receiver, &rl->key, 1, where, 1, count, use_nested_array, reply_nil_when_empty, &deleted);
updateStatsOnUnblock(receiver, 0, elapsedUs(replyTimer)); updateStatsOnUnblock(receiver, 0, elapsedUs(replyTimer));
unblockClient(receiver); unblockClient(receiver);
zcard--;
/* Replicate the command. */ /* Replicate the command. */
robj *argv[2]; int argc = 2;
robj *argv[3];
argv[0] = where == ZSET_MIN ? shared.zpopmin : shared.zpopmax; argv[0] = where == ZSET_MIN ? shared.zpopmin : shared.zpopmax;
argv[1] = rl->key; argv[1] = rl->key;
incrRefCount(rl->key); incrRefCount(rl->key);
propagate(receiver->db->id,argv,2,PROPAGATE_AOF|PROPAGATE_REPL); if (count != 0) {
/* Replicate it as command with COUNT. */
robj *count_obj = createStringObjectFromLongLong((count > llen) ? llen : count);
argv[2] = count_obj;
argc++;
}
propagate(receiver->db->id, argv, argc, PROPAGATE_AOF|PROPAGATE_REPL);
decrRefCount(argv[1]); decrRefCount(argv[1]);
if (count != 0) decrRefCount(argv[2]);
/* The zset is empty and has been deleted. */
if (deleted) break;
} }
} }
} }
...@@ -613,7 +628,7 @@ void handleClientsBlockedOnKeys(void) { ...@@ -613,7 +628,7 @@ void handleClientsBlockedOnKeys(void) {
* *
* 'count' for those commands that support the optional count argument. * 'count' for those commands that support the optional count argument.
* Otherwise the value is 0. */ * 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) { void blockForKeys(client *c, int btype, robj **keys, int numkeys, long count, mstime_t timeout, robj *target, struct blockPos *blockpos, streamID *ids) {
dictEntry *de; dictEntry *de;
list *l; list *l;
int j; int j;
...@@ -622,7 +637,7 @@ void blockForKeys(client *c, int btype, robj **keys, int numkeys, long count, ms ...@@ -622,7 +637,7 @@ void blockForKeys(client *c, int btype, robj **keys, int numkeys, long count, ms
c->bpop.timeout = timeout; c->bpop.timeout = timeout;
c->bpop.target = target; c->bpop.target = target;
if (listpos != NULL) c->bpop.listpos = *listpos; if (blockpos != NULL) c->bpop.blockpos = *blockpos;
if (target != NULL) incrRefCount(target); if (target != NULL) incrRefCount(target);
......
...@@ -1711,6 +1711,16 @@ int blmpopGetKeys(struct redisCommand *cmd, robj **argv, int argc, getKeysResult ...@@ -1711,6 +1711,16 @@ int blmpopGetKeys(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 zmpopGetKeys(struct redisCommand *cmd, robj **argv, int argc, getKeysResult *result) {
UNUSED(cmd);
return genericGetKeys(0, 1, 2, 1, argv, argc, result);
}
int bzmpopGetKeys(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> ...
......
...@@ -792,6 +792,13 @@ struct redisCommand redisCommandTable[] = { ...@@ -792,6 +792,13 @@ struct redisCommand redisCommandTable[] = {
KSPEC_BS_INDEX,.bs.index={1}, KSPEC_BS_INDEX,.bs.index={1},
KSPEC_FK_RANGE,.fk.range={0,1,0}}}}, KSPEC_FK_RANGE,.fk.range={0,1,0}}}},
{"zmpop", zmpopCommand,-4,
"write @sortedset",
{{"write",
KSPEC_BS_INDEX,.bs.index={1},
KSPEC_FK_KEYNUM,.fk.keynum={0,1,1}}},
zmpopGetKeys},
{"bzpopmin",bzpopminCommand,-3, {"bzpopmin",bzpopminCommand,-3,
"write no-script fast @sortedset @blocking", "write no-script fast @sortedset @blocking",
{{"write", {{"write",
...@@ -804,6 +811,13 @@ struct redisCommand redisCommandTable[] = { ...@@ -804,6 +811,13 @@ struct redisCommand redisCommandTable[] = {
KSPEC_BS_INDEX,.bs.index={1}, KSPEC_BS_INDEX,.bs.index={1},
KSPEC_FK_RANGE,.fk.range={-2,1,0}}}}, KSPEC_FK_RANGE,.fk.range={-2,1,0}}}},
{"bzmpop",bzmpopCommand,-5,
"write @sortedset @blocking",
{{"write",
KSPEC_BS_INDEX,.bs.index={2},
KSPEC_FK_KEYNUM,.fk.keynum={0,1,1}}},
blmpopGetKeys},
{"zrandmember",zrandmemberCommand,-2, {"zrandmember",zrandmemberCommand,-2,
"read-only random @sortedset", "read-only random @sortedset",
{{"read", {{"read",
......
...@@ -812,12 +812,12 @@ typedef struct blockingState { ...@@ -812,12 +812,12 @@ typedef struct blockingState {
* operation such as BLPOP or XREAD. Or NULL. */ * operation such as BLPOP or XREAD. Or NULL. */
robj *target; /* The key that should receive the element, robj *target; /* The key that should receive the element,
* for BLMOVE. */ * for BLMOVE. */
struct listPos { struct blockPos {
int wherefrom; /* Where to pop from */ int wherefrom; /* Where to pop from */
int whereto; /* Where to push to */ int whereto; /* Where to push to */
} listpos; /* The positions in the src/dst lists } blockpos; /* The positions in the src/dst lists/zsets
* where we want to pop/push an element * where we want to pop/push an element
* for BLPOP, BRPOP and BLMOVE. */ * for BLPOP, BRPOP, BLMOVE and BZMPOP. */
/* BLOCK_STREAM */ /* BLOCK_STREAM */
size_t xread_count; /* XREAD COUNT option. */ size_t xread_count; /* XREAD COUNT option. */
...@@ -2343,7 +2343,7 @@ int zsetAdd(robj *zobj, double score, sds ele, int in_flags, int *out_flags, dou ...@@ -2343,7 +2343,7 @@ int zsetAdd(robj *zobj, double score, sds ele, int in_flags, int *out_flags, dou
long zsetRank(robj *zobj, sds ele, int reverse); long zsetRank(robj *zobj, sds ele, int reverse);
int zsetDel(robj *zobj, sds ele); int zsetDel(robj *zobj, sds ele);
robj *zsetDup(robj *o); robj *zsetDup(robj *o);
void genericZpopCommand(client *c, robj **keyv, int keyc, int where, int emitkey, robj *countarg); void genericZpopCommand(client *c, robj **keyv, int keyc, int where, int emitkey, long count, int use_nested_array, int reply_nil_when_empty, int *deleted);
sds lpGetObject(unsigned char *sptr); sds lpGetObject(unsigned char *sptr);
int zslValueGteMin(double value, zrangespec *spec); int zslValueGteMin(double value, zrangespec *spec);
int zslValueLteMax(double value, zrangespec *spec); int zslValueLteMax(double value, zrangespec *spec);
...@@ -2557,6 +2557,8 @@ int memoryGetKeys(struct redisCommand *cmd, robj **argv, int argc, getKeysResult ...@@ -2557,6 +2557,8 @@ int memoryGetKeys(struct redisCommand *cmd, robj **argv, int argc, getKeysResult
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 lmpopGetKeys(struct redisCommand *cmd, robj **argv, int argc, getKeysResult *result);
int blmpopGetKeys(struct redisCommand *cmd, robj **argv, int argc, getKeysResult *result); int blmpopGetKeys(struct redisCommand *cmd, robj **argv, int argc, getKeysResult *result);
int zmpopGetKeys(struct redisCommand *cmd, robj **argv, int argc, getKeysResult *result);
int bzmpopGetKeys(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);
...@@ -2593,7 +2595,7 @@ int getTimeoutFromObjectOrReply(client *c, robj *object, mstime_t *timeout, int ...@@ -2593,7 +2595,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, long count, 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 blockPos *blockpos, 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. */
...@@ -2751,8 +2753,10 @@ void zremrangebyscoreCommand(client *c); ...@@ -2751,8 +2753,10 @@ void zremrangebyscoreCommand(client *c);
void zremrangebylexCommand(client *c); void zremrangebylexCommand(client *c);
void zpopminCommand(client *c); void zpopminCommand(client *c);
void zpopmaxCommand(client *c); void zpopmaxCommand(client *c);
void zmpopCommand(client *c);
void bzpopminCommand(client *c); void bzpopminCommand(client *c);
void bzpopmaxCommand(client *c); void bzpopmaxCommand(client *c);
void bzmpopCommand(client *c);
void zrandmemberCommand(client *c); void zrandmemberCommand(client *c);
void multiCommand(client *c); void multiCommand(client *c);
void execCommand(client *c); void execCommand(client *c);
......
...@@ -1104,7 +1104,7 @@ void blockingPopGenericCommand(client *c, robj **keys, int numkeys, int where, i ...@@ -1104,7 +1104,7 @@ void blockingPopGenericCommand(client *c, robj **keys, int numkeys, int where, i
} }
/* If the keys do not exist we must block */ /* If the keys do not exist we must block */
struct listPos pos = {where}; struct blockPos pos = {where};
blockForKeys(c,BLOCKED_LIST,keys,numkeys,count,timeout,NULL,&pos,NULL); blockForKeys(c,BLOCKED_LIST,keys,numkeys,count,timeout,NULL,&pos,NULL);
} }
...@@ -1129,7 +1129,7 @@ void blmoveGenericCommand(client *c, int wherefrom, int whereto, mstime_t timeou ...@@ -1129,7 +1129,7 @@ void blmoveGenericCommand(client *c, int wherefrom, int whereto, mstime_t timeou
addReplyNull(c); addReplyNull(c);
} else { } else {
/* The list is empty and the client blocks. */ /* The list is empty and the client blocks. */
struct listPos pos = {wherefrom, whereto}; struct blockPos pos = {wherefrom, whereto};
blockForKeys(c,BLOCKED_LIST,c->argv + 1,1,0,timeout,c->argv[2],&pos,NULL); blockForKeys(c,BLOCKED_LIST,c->argv + 1,1,0,timeout,c->argv[2],&pos,NULL);
} }
} else { } else {
......
...@@ -3753,31 +3753,35 @@ void zscanCommand(client *c) { ...@@ -3753,31 +3753,35 @@ void zscanCommand(client *c) {
} }
/* This command implements the generic zpop operation, used by: /* This command implements the generic zpop operation, used by:
* ZPOPMIN, ZPOPMAX, BZPOPMIN and BZPOPMAX. This function is also used * ZPOPMIN, ZPOPMAX, BZPOPMIN, BZPOPMAX and ZMPOP. This function is also used
* inside blocked.c in the unblocking stage of BZPOPMIN and BZPOPMAX. * inside blocked.c in the unblocking stage of BZPOPMIN, BZPOPMAX and BZMPOP.
* *
* If 'emitkey' is true also the key name is emitted, useful for the blocking * If 'emitkey' is true also the key name is emitted, useful for the blocking
* behavior of BZPOP[MIN|MAX], since we can block into multiple keys. * behavior of BZPOP[MIN|MAX], since we can block into multiple keys.
* Or in ZMPOP/BZMPOP, because we also can take multiple keys.
* *
* The synchronous version instead does not need to emit the key, but may * 'count' is the number of elements requested to pop, or 0 for plain single pop.
* use the 'count' argument to return multiple items if available. */ *
void genericZpopCommand(client *c, robj **keyv, int keyc, int where, int emitkey, robj *countarg) { * 'use_nested_array' when false it generates a flat array (with or without key name).
* When true, it generates a nested 2 level array of field + score pairs, or 3 level when emitkey is set.
*
* 'reply_nil_when_empty' when true we reply a NIL if we are not able to pop up any elements.
* Like in ZMPOP/BZMPOP we reply with a structured nested array containing key name
* and member + score pairs. In these commands, we reply with null when we have no result.
* Otherwise in ZPOPMIN/ZPOPMAX we reply an empty array by default.
*
* 'deleted' is an optional output argument to get an indication
* if the key got deleted by this function.
* */
void genericZpopCommand(client *c, robj **keyv, int keyc, int where, int emitkey,
long count, int use_nested_array, int reply_nil_when_empty, int *deleted) {
int idx; int idx;
robj *key = NULL; robj *key = NULL;
robj *zobj = NULL; robj *zobj = NULL;
sds ele; sds ele;
double score; double score;
long count = 1;
/* If a count argument as passed, parse it or return an error. */ if (deleted) *deleted = 0;
if (countarg) {
if (getLongFromObjectOrReply(c,countarg,&count,NULL) != C_OK)
return;
if (count <= 0) {
addReply(c,shared.emptyarray);
return;
}
}
/* Check type and break on the first error, otherwise identify candidate. */ /* Check type and break on the first error, otherwise identify candidate. */
idx = 0; idx = 0;
...@@ -3791,20 +3795,38 @@ void genericZpopCommand(client *c, robj **keyv, int keyc, int where, int emitkey ...@@ -3791,20 +3795,38 @@ void genericZpopCommand(client *c, robj **keyv, int keyc, int where, int emitkey
/* No candidate for zpopping, return empty. */ /* No candidate for zpopping, return empty. */
if (!zobj) { if (!zobj) {
addReply(c,shared.emptyarray); if (reply_nil_when_empty) {
addReplyNullArray(c);
} else {
addReply(c,shared.emptyarray);
}
return; return;
} }
void *arraylen_ptr = addReplyDeferredLen(c);
long result_count = 0; long result_count = 0;
/* We emit the key only for the blocking variant. */ /* When count is 0, we need to correct it to 1 for plain single pop. */
if (emitkey) addReplyBulk(c,key); if (count == 0) count = 1;
/* Respond with a single (flat) array in RESP2 or if countarg is not long llen = zsetLength(zobj);
* provided (returning a single element). In RESP3, when countarg is long rangelen = (count > llen) ? llen : count;
* provided, use nested array. */
int use_nested_array = c->resp > 2 && countarg != NULL; if (!use_nested_array && !emitkey) {
/* ZPOPMIN/ZPOPMAX with or without COUNT option in RESP2. */
addReplyArrayLen(c, rangelen * 2);
} else if (use_nested_array && !emitkey) {
/* ZPOPMIN/ZPOPMAX with COUNT option in RESP3. */
addReplyArrayLen(c, rangelen);
} else if (!use_nested_array && emitkey) {
/* BZPOPMIN/BZPOPMAX in RESP2 and RESP3. */
addReplyArrayLen(c, rangelen * 2 + 1);
addReplyBulk(c, key);
} else if (use_nested_array && emitkey) {
/* ZMPOP/BZMPOP in RESP2 and RESP3. */
addReplyArrayLen(c, 2);
addReplyBulk(c, key);
addReplyArrayLen(c, rangelen);
}
/* Remove the element. */ /* Remove the element. */
do { do {
...@@ -3861,64 +3883,114 @@ void genericZpopCommand(client *c, robj **keyv, int keyc, int where, int emitkey ...@@ -3861,64 +3883,114 @@ void genericZpopCommand(client *c, robj **keyv, int keyc, int where, int emitkey
addReplyDouble(c,score); addReplyDouble(c,score);
sdsfree(ele); sdsfree(ele);
++result_count; ++result_count;
} while(--rangelen);
/* Remove the key, if indeed needed. */ /* Remove the key, if indeed needed. */
if (zsetLength(zobj) == 0) { if (zsetLength(zobj) == 0) {
dbDelete(c->db,key); if (deleted) *deleted = 1;
notifyKeyspaceEvent(NOTIFY_GENERIC,"del",key,c->db->id);
break;
}
} while(--count);
if (!use_nested_array) { dbDelete(c->db,key);
result_count *= 2; notifyKeyspaceEvent(NOTIFY_GENERIC,"del",key,c->db->id);
}
if (c->cmd->proc == zmpopCommand) {
/* Always replicate it as ZPOP[MIN|MAX] with COUNT option instead of ZMPOP. */
robj *count_obj = createStringObjectFromLongLong((count > llen) ? llen : count);
rewriteClientCommandVector(c, 3,
(where == ZSET_MAX) ? shared.zpopmax : shared.zpopmin,
key, count_obj);
decrRefCount(count_obj);
} }
setDeferredArrayLen(c,arraylen_ptr,result_count + (emitkey != 0));
} }
/* ZPOPMIN key [<count>] */ /* ZPOPMIN/ZPOPMAX key [<count>] */
void zpopminCommand(client *c) { void zpopMinMaxCommand(client *c, int where) {
if (c->argc > 3) { if (c->argc > 3) {
addReplyErrorObject(c,shared.syntaxerr); addReplyErrorObject(c,shared.syntaxerr);
return; return;
} }
genericZpopCommand(c,&c->argv[1],1,ZSET_MIN,0,
c->argc == 3 ? c->argv[2] : NULL); long count = 0; /* 0 for plain single pop. */
if (c->argc == 3) {
if (getLongFromObjectOrReply(c, c->argv[2], &count, NULL) != C_OK)
return;
if (count <= 0) {
addReply(c, shared.emptyarray);
return;
}
}
/* Respond with a single (flat) array in RESP2 or if count is 0
* (returning a single element). In RESP3, when count > 0 use nested array. */
int use_nested_array = (c->resp > 2 && count != 0);
genericZpopCommand(c, &c->argv[1], 1, where, 0, count, use_nested_array, 0, NULL);
}
/* ZPOPMIN key [<count>] */
void zpopminCommand(client *c) {
zpopMinMaxCommand(c, ZSET_MIN);
} }
/* ZMAXPOP key [<count>] */ /* ZMAXPOP key [<count>] */
void zpopmaxCommand(client *c) { void zpopmaxCommand(client *c) {
if (c->argc > 3) { zpopMinMaxCommand(c, ZSET_MAX);
addReplyErrorObject(c,shared.syntaxerr);
return;
}
genericZpopCommand(c,&c->argv[1],1,ZSET_MAX,0,
c->argc == 3 ? c->argv[2] : NULL);
} }
/* BZPOPMIN / BZPOPMAX actual implementation. */ /* BZPOPMIN, BZPOPMAX, BZMPOP actual implementation.
void blockingGenericZpopCommand(client *c, int where) { *
* 'numkeys' is the number of keys.
*
* 'timeout_idx' parameter position of block timeout.
*
* 'where' ZSET_MIN or ZSET_MAX.
*
* 'count' is the number of elements requested to pop, or 0 for plain single pop.
*
* 'use_nested_array' when false it generates a flat array (with or without key name).
* When true, it generates a nested 3 level array of keyname, field + score pairs.
* */
void blockingGenericZpopCommand(client *c, robj **keys, int numkeys, int where,
int timeout_idx, long count, int use_nested_array, int reply_nil_when_empty) {
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++) { for (j = 0; j < numkeys; j++) {
o = lookupKeyWrite(c->db,c->argv[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_ZSET)) return; if (checkType(c,o,OBJ_ZSET)) return;
if (o != NULL) {
if (zsetLength(o) != 0) { long llen = zsetLength(o);
/* Non empty zset, this is like a normal ZPOP[MIN|MAX]. */ /* Empty zset, move to next key. */
genericZpopCommand(c,&c->argv[j],1,where,1,NULL); if (llen == 0) continue;
/* Replicate it as an ZPOP[MIN|MAX] instead of BZPOP[MIN|MAX]. */
rewriteClientCommandVector(c,2, /* Non empty zset, this is like a normal ZPOP[MIN|MAX]. */
where == ZSET_MAX ? shared.zpopmax : shared.zpopmin, genericZpopCommand(c, &key, 1, where, 1, count, use_nested_array, reply_nil_when_empty, NULL);
c->argv[j]);
return; if (count == 0) {
} /* Replicate it as ZPOP[MIN|MAX] instead of BZPOP[MIN|MAX]. */
rewriteClientCommandVector(c,2,
(where == ZSET_MAX) ? shared.zpopmax : shared.zpopmin,
key);
} else {
/* Replicate it as ZPOP[MIN|MAX] with COUNT option. */
robj *count_obj = createStringObjectFromLongLong((count > llen) ? llen : count);
rewriteClientCommandVector(c, 3,
(where == ZSET_MAX) ? shared.zpopmax : shared.zpopmin,
key, count_obj);
decrRefCount(count_obj);
} }
return;
} }
/* If we are not allowed to block the client and the zset is empty the only thing /* If we are not allowed to block the client and the zset is empty the only thing
...@@ -3929,17 +4001,18 @@ void blockingGenericZpopCommand(client *c, int where) { ...@@ -3929,17 +4001,18 @@ 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,0,timeout,NULL,NULL,NULL); struct blockPos pos = {where};
blockForKeys(c,BLOCKED_ZSET,c->argv+1,c->argc-2,count,timeout,NULL,&pos,NULL);
} }
// BZPOPMIN key [key ...] timeout // BZPOPMIN key [key ...] timeout
void bzpopminCommand(client *c) { void bzpopminCommand(client *c) {
blockingGenericZpopCommand(c,ZSET_MIN); blockingGenericZpopCommand(c, c->argv+1, c->argc-2, ZSET_MIN, c->argc-1, 0, 0, 0);
} }
// BZPOPMAX key [key ...] timeout // BZPOPMAX key [key ...] timeout
void bzpopmaxCommand(client *c) { void bzpopmaxCommand(client *c) {
blockingGenericZpopCommand(c,ZSET_MAX); blockingGenericZpopCommand(c, c->argv+1, c->argc-2, ZSET_MAX, c->argc-1, 0, 0, 0);
} }
static void zarndmemberReplyWithListpack(client *c, unsigned int count, listpackEntry *keys, listpackEntry *vals) { static void zarndmemberReplyWithListpack(client *c, unsigned int count, listpackEntry *keys, listpackEntry *vals) {
...@@ -4189,3 +4262,68 @@ void zrandmemberCommand(client *c) { ...@@ -4189,3 +4262,68 @@ void zrandmemberCommand(client *c) {
zsetTypeRandomElement(zset, zsetLength(zset), &ele,NULL); zsetTypeRandomElement(zset, zsetLength(zset), &ele,NULL);
zsetReplyFromListpackEntry(c,&ele); zsetReplyFromListpackEntry(c,&ele);
} }
/* ZMPOP/BZMPOP
*
* 'numkeys_idx' parameter position of key number.
* 'is_block' this indicates whether it is a blocking variant. */
void zmpopGenericCommand(client *c, int numkeys_idx, int is_block) {
long j;
long numkeys = 0; /* Number of keys. */
int where = 0; /* ZSET_MIN or ZSET_MAX. */
long count = 1; /* Reply will consist of up to count elements, depending on the zset'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 (!strcasecmp(c->argv[where_idx]->ptr, "MIN")) {
where = ZSET_MIN;
} else if (!strcasecmp(c->argv[where_idx]->ptr, "MAX")) {
where = ZSET_MAX;
} else {
addReplyErrorObject(c, shared.syntaxerr);
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 blockingGenericZpopCommand. */
blockingGenericZpopCommand(c, c->argv+numkeys_idx+1, numkeys, where, 1, count, 1, 1);
} else {
/* NON-BLOCK */
genericZpopCommand(c, c->argv+numkeys_idx+1, numkeys, where, 1, count, 1, 1, NULL);
}
}
/* ZMPOP numkeys [<key> ...] MIN|MAX [COUNT count] */
void zmpopCommand(client *c) {
zmpopGenericCommand(c, 1, 0);
}
/* BZMPOP timeout numkeys [<key> ...] MIN|MAX [COUNT count] */
void bzmpopCommand(client *c) {
zmpopGenericCommand(c, 2, 1);
}
...@@ -333,7 +333,6 @@ tags {"aof external:skip"} { ...@@ -333,7 +333,6 @@ tags {"aof external:skip"} {
set client [redis [dict get $srv host] [dict get $srv port] 0 $::tls] 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] set client2 [redis [dict get $srv host] [dict get $srv port] 1 $::tls]
wait_done_loading $client wait_done_loading $client
wait_done_loading $client2
# Pop all elements from mylist, should be blmpop delete mylist. # Pop all elements from mylist, should be blmpop delete mylist.
$client lmpop 1 mylist left count 1 $client lmpop 1 mylist left count 1
...@@ -368,4 +367,51 @@ tags {"aof external:skip"} { ...@@ -368,4 +367,51 @@ tags {"aof external:skip"} {
assert_equal 2 [$client llen mylist3] assert_equal 2 [$client llen mylist3]
} }
} }
# Test that ZMPOP/BZMPOP work fine with AOF.
create_aof {
append_to_aof [formatCommand zadd myzset 1 one 2 two 3 three]
append_to_aof [formatCommand zadd myzset2 4 four 5 five 6 six]
append_to_aof [formatCommand zadd myzset3 1 one 2 two 3 three 4 four 5 five]
}
start_server_aof [list dir $server_path aof-load-truncated no] {
test "AOF+ZMPOP/BZMPOP: pop elements from the zset" {
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
# Pop all elements from myzset, should be bzmpop delete myzset.
$client zmpop 1 myzset min count 1
$client bzmpop 0 1 myzset min count 10
# Pop all elements from myzset2, should be zmpop delete myzset2.
$client bzmpop 0 2 myzset myzset2 max count 10
$client zmpop 2 myzset myzset2 max count 2
# Blocking path, be blocked and then released.
$client2 bzmpop 0 2 myzset myzset2 min count 2
after 100
$client zadd myzset2 1 one 2 two 3 three
# Pop up the last element in myzset2
$client bzmpop 0 3 myzset myzset2 myzset3 min count 1
# Leave two elements in myzset3.
$client bzmpop 0 3 myzset myzset2 myzset3 max count 3
}
}
start_server_aof [list dir $server_path aof-load-truncated no] {
test "AOF+ZMPOP/BZMPOP: after pop elements from the zset" {
set client [redis [dict get $srv host] [dict get $srv port] 0 $::tls]
wait_done_loading $client
# myzset and myzset2 no longer exist.
assert_equal 0 [$client exists myzset myzset2]
# Length of myzset3 is two.
assert_equal 2 [$client zcard myzset3]
}
}
} }
...@@ -6,7 +6,7 @@ start_server { ...@@ -6,7 +6,7 @@ start_server {
} { } {
source "tests/unit/type/list-common.tcl" source "tests/unit/type/list-common.tcl"
# A helper function for BPOP/BLMPOP with one input key. # A helper function to execute either B*POP or BLMPOP* with one input key.
proc bpop_command {rd pop key timeout} { proc bpop_command {rd pop key timeout} {
if {$pop == "BLMPOP_LEFT"} { if {$pop == "BLMPOP_LEFT"} {
$rd blmpop $timeout 1 $key left count 1 $rd blmpop $timeout 1 $key left count 1
...@@ -17,7 +17,7 @@ start_server { ...@@ -17,7 +17,7 @@ start_server {
} }
} }
# A helper function for BPOP/BLMPOP with two input keys. # A helper function to execute either B*POP or BLMPOP* with two input keys.
proc bpop_command_two_key {rd pop key key2 timeout} { proc bpop_command_two_key {rd pop key key2 timeout} {
if {$pop == "BLMPOP_LEFT"} { if {$pop == "BLMPOP_LEFT"} {
$rd blmpop $timeout 2 $key $key2 left count 1 $rd blmpop $timeout 2 $key $key2 left count 1
...@@ -719,14 +719,14 @@ foreach {pop} {BLPOP BLMPOP_LEFT} { ...@@ -719,14 +719,14 @@ foreach {pop} {BLPOP BLMPOP_LEFT} {
set rd [redis_deferring_client] set rd [redis_deferring_client]
set repl [attach_to_replication_stream] set repl [attach_to_replication_stream]
# BLMPOP without block. # BLMPOP without being blocked.
r lpush mylist{t} a b c r lpush mylist{t} a b c
r rpush mylist2{t} 1 2 3 r rpush mylist2{t} 1 2 3
r blmpop 0 1 mylist{t} left count 1 r blmpop 0 1 mylist{t} left count 1
r blmpop 0 2 mylist{t} mylist2{t} right count 10 r blmpop 0 2 mylist{t} mylist2{t} right count 10
r blmpop 0 2 mylist{t} mylist2{t} right count 10 r blmpop 0 2 mylist{t} mylist2{t} right count 10
# BLMPOP with block. # BLMPOP that gets blocked.
$rd blmpop 0 1 mylist{t} left count 1 $rd blmpop 0 1 mylist{t} left count 1
wait_for_blocked_client wait_for_blocked_client
r lpush mylist{t} a r lpush mylist{t} a
...@@ -737,6 +737,10 @@ foreach {pop} {BLPOP BLMPOP_LEFT} { ...@@ -737,6 +737,10 @@ foreach {pop} {BLPOP BLMPOP_LEFT} {
wait_for_blocked_client wait_for_blocked_client
r rpush mylist2{t} a b c r rpush mylist2{t} a b c
# Released on timeout.
assert_equal {} [r blmpop 0.01 1 mylist{t} left count 10]
r set foo{t} bar ;# something else to propagate after, so we can make sure the above pop didn't.
assert_replication_stream $repl { assert_replication_stream $repl {
{select *} {select *}
{lpush mylist{t} a b c} {lpush mylist{t} a b c}
...@@ -750,6 +754,7 @@ foreach {pop} {BLPOP BLMPOP_LEFT} { ...@@ -750,6 +754,7 @@ foreach {pop} {BLPOP BLMPOP_LEFT} {
{lpop mylist{t} 3} {lpop mylist{t} 3}
{rpush mylist2{t} a b c} {rpush mylist2{t} a b c}
{rpop mylist2{t} 3} {rpop mylist2{t} 3}
{set foo{t} bar}
} }
} {} {needs:repl} } {} {needs:repl}
......
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