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

Fix XREADGROUP BLOCK stuck in endless loop (#12301)



For the XREADGROUP BLOCK > scenario, there is an endless loop.
Due to #11012, it keep going, reprocess command -> blockForKeys -> reprocess command

The right fix is to avoid an endless loop in handleClientsBlockedOnKey and handleClientsBlockedOnKeys,
looks like there was some attempt in handleClientsBlockedOnKeys but maybe not sufficiently good,
and it looks like using a similar trick in handleClientsBlockedOnKey is complicated.
i.e. stashing the list on the stack and iterating on it after creating a fresh one for future use,
is problematic since the code keeps accessing the global list.
Co-authored-by: default avatarOran Agra <oran@redislabs.com>
parent f228ec1e
...@@ -564,7 +564,10 @@ static void handleClientsBlockedOnKey(readyList *rl) { ...@@ -564,7 +564,10 @@ static void handleClientsBlockedOnKey(readyList *rl) {
listIter li; listIter li;
listRewind(clients,&li); listRewind(clients,&li);
while((ln = listNext(&li))) { /* Avoid processing more than the initial count so that we're not stuck
* in an endless loop in case the reprocessing of the command blocks again. */
long count = listLength(clients);
while ((ln = listNext(&li)) && count--) {
client *receiver = listNodeValue(ln); client *receiver = listNodeValue(ln);
robj *o = lookupKeyReadWithFlags(rl->db, rl->key, LOOKUP_NOEFFECTS); robj *o = lookupKeyReadWithFlags(rl->db, rl->key, LOOKUP_NOEFFECTS);
/* 1. In case new key was added/touched we need to verify it satisfy the /* 1. In case new key was added/touched we need to verify it satisfy the
......
...@@ -474,7 +474,30 @@ start_server { ...@@ -474,7 +474,30 @@ start_server {
$rd close $rd close
} }
test {Blocking XREADGROUP for stream key that has clients blocked on list - avoid endless loop} {
r DEL mystream
r XGROUP CREATE mystream mygroup $ MKSTREAM
set rd1 [redis_deferring_client]
set rd2 [redis_deferring_client]
set rd3 [redis_deferring_client]
$rd1 xreadgroup GROUP mygroup myuser COUNT 10 BLOCK 10000 STREAMS mystream >
$rd2 xreadgroup GROUP mygroup myuser COUNT 10 BLOCK 10000 STREAMS mystream >
$rd3 xreadgroup GROUP mygroup myuser COUNT 10 BLOCK 10000 STREAMS mystream >
wait_for_blocked_clients_count 3
r xadd mystream MAXLEN 5000 * field1 value1 field2 value2 field3 value3
$rd1 close
$rd2 close
$rd3 close
assert_equal [r ping] {PONG}
}
test {XGROUP DESTROY should unblock XREADGROUP with -NOGROUP} { test {XGROUP DESTROY should unblock XREADGROUP with -NOGROUP} {
r config resetstat r config resetstat
r del mystream r del mystream
......
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