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

Fix WAITAOF reply when using last_offset and last_numreplicas (#11917)

WAITAOF wad added in #11713, its return is an array.
But forget to handle WAITAOF in last_offset and last_numreplicas,
causing WAITAOF to return a WAIT like reply.

Tests was added to validate that case (both WAIT and WAITAOF).
This PR also refactored processClientsWaitingReplicas a bit for better
maintainability and readability.
parent 5360350e
...@@ -3575,7 +3575,7 @@ void waitaofCommand(client *c) { ...@@ -3575,7 +3575,7 @@ void waitaofCommand(client *c) {
return; return;
} }
if (numlocal && !server.aof_enabled) { if (numlocal && !server.aof_enabled) {
addReplyError(c,"WAITAOF cannot be used when appendonly is disabled"); addReplyError(c, "WAITAOF cannot be used when numlocal is set but appendonly is disabled.");
return; return;
} }
...@@ -3619,51 +3619,56 @@ void processClientsWaitingReplicas(void) { ...@@ -3619,51 +3619,56 @@ void processClientsWaitingReplicas(void) {
listRewind(server.clients_waiting_acks,&li); listRewind(server.clients_waiting_acks,&li);
while((ln = listNext(&li))) { while((ln = listNext(&li))) {
int numlocal = 0;
int numreplicas = 0;
client *c = ln->value; client *c = ln->value;
int is_wait_aof = c->bstate.btype == BLOCKED_WAITAOF; int is_wait_aof = c->bstate.btype == BLOCKED_WAITAOF;
if (is_wait_aof && c->bstate.numlocal && !server.aof_enabled) { if (is_wait_aof && c->bstate.numlocal && !server.aof_enabled) {
addReplyError(c, "WAITAOF cannot be used when numlocal is set but appendonly is disabled.");
unblockClient(c); unblockClient(c);
addReplyError(c,"WAITAOF cannot be used when appendonly is disabled");
return; return;
} }
/* Every time we find a client that is satisfied for a given /* Every time we find a client that is satisfied for a given
* offset and number of replicas, we remember it so the next client * offset and number of replicas, we remember it so the next client
* may be unblocked without calling replicationCountAcksByOffset() * may be unblocked without calling replicationCountAcksByOffset()
* or calling replicationCountAOFAcksByOffset()
* if the requested offset / replicas were equal or less. */ * if the requested offset / replicas were equal or less. */
if (last_offset && last_offset >= c->bstate.reploffset && if (last_offset && last_offset >= c->bstate.reploffset &&
last_numreplicas >= c->bstate.numreplicas) last_numreplicas >= c->bstate.numreplicas)
{ {
/* Reply before unblocking, because unblock client calls reqresAppendResponse */ numreplicas = last_numreplicas;
addReplyLongLong(c,last_numreplicas);
unblockClient(c);
} else { } else {
int numreplicas = is_wait_aof ? numreplicas = is_wait_aof ?
replicationCountAOFAcksByOffset(c->bstate.reploffset) : replicationCountAOFAcksByOffset(c->bstate.reploffset) :
replicationCountAcksByOffset(c->bstate.reploffset); replicationCountAcksByOffset(c->bstate.reploffset);
if (numreplicas >= c->bstate.numreplicas) { /* Check if the number of replicas is satisfied. */
last_offset = c->bstate.reploffset; if (numreplicas < c->bstate.numreplicas) continue;
last_numreplicas = numreplicas;
/* Check if the local constraint of WAITAOF is served */ last_offset = c->bstate.reploffset;
int numlocal = server.fsynced_reploff >= c->bstate.reploffset; last_numreplicas = numreplicas;
if (is_wait_aof && numlocal < c->bstate.numlocal) }
continue;
if (is_wait_aof) { /* Check if the local constraint of WAITAOF is served */
/* WAITAOF has an array reply*/ if (is_wait_aof) {
addReplyArrayLen(c,2); numlocal = server.fsynced_reploff >= c->bstate.reploffset;
addReplyLongLong(c,numlocal); if (numlocal < c->bstate.numlocal) continue;
addReplyLongLong(c,numreplicas);
} else {
addReplyLongLong(c,numreplicas);
}
/* Reply before unblocking, because unblock client calls reqresAppendResponse */
unblockClient(c);
}
} }
/* Reply before unblocking, because unblock client calls reqresAppendResponse */
if (is_wait_aof) {
/* WAITAOF has an array reply */
addReplyArrayLen(c, 2);
addReplyLongLong(c, numlocal);
addReplyLongLong(c, numreplicas);
} else {
addReplyLongLong(c, numreplicas);
}
unblockClient(c);
} }
} }
......
...@@ -886,9 +886,9 @@ proc wait_for_blocked_client {{idx 0}} { ...@@ -886,9 +886,9 @@ proc wait_for_blocked_client {{idx 0}} {
} }
} }
proc wait_for_blocked_clients_count {count {maxtries 100} {delay 10}} { proc wait_for_blocked_clients_count {count {maxtries 100} {delay 10} {idx 0}} {
wait_for_condition $maxtries $delay { wait_for_condition $maxtries $delay {
[s blocked_clients] == $count [s $idx blocked_clients] == $count
} else { } else {
fail "Timeout waiting for blocked clients" fail "Timeout waiting for blocked clients"
} }
......
...@@ -18,20 +18,20 @@ start_server {} { ...@@ -18,20 +18,20 @@ start_server {} {
fail "Replication not started." fail "Replication not started."
} }
} }
test {WAIT out of range timeout (milliseconds)} { test {WAIT out of range timeout (milliseconds)} {
# Timeout is parsed as milliseconds by getLongLongFromObjectOrReply(). # Timeout is parsed as milliseconds by getLongLongFromObjectOrReply().
# Verify we get out of range message if value is behind LLONG_MAX # Verify we get out of range message if value is behind LLONG_MAX
# (decimal value equals to 0x8000000000000000) # (decimal value equals to 0x8000000000000000)
assert_error "*or out of range*" {$master wait 2 9223372036854775808} assert_error "*or out of range*" {$master wait 2 9223372036854775808}
# expected to fail by later overflow condition after addition # expected to fail by later overflow condition after addition
# of mstime(). (decimal value equals to 0x7FFFFFFFFFFFFFFF) # of mstime(). (decimal value equals to 0x7FFFFFFFFFFFFFFF)
assert_error "*timeout is out of range*" {$master wait 2 9223372036854775807} assert_error "*timeout is out of range*" {$master wait 2 9223372036854775807}
assert_error "*timeout is negative*" {$master wait 2 -1} assert_error "*timeout is negative*" {$master wait 2 -1}
} }
test {WAIT should acknowledge 1 additional copy of the data} { test {WAIT should acknowledge 1 additional copy of the data} {
$master set foo 0 $master set foo 0
$master incr foo $master incr foo
...@@ -68,6 +68,36 @@ start_server {} { ...@@ -68,6 +68,36 @@ start_server {} {
exec kill -SIGCONT $slave_pid exec kill -SIGCONT $slave_pid
assert {[$master wait 1 1000] == 1} assert {[$master wait 1 1000] == 1}
} }
test {WAIT replica multiple clients unblock - reuse last result} {
set rd [redis_deferring_client -1]
set rd2 [redis_deferring_client -1]
exec kill -SIGSTOP $slave_pid
$rd incr foo
$rd read
$rd2 incr foo
$rd2 read
$rd wait 1 0
$rd2 wait 1 0
wait_for_blocked_clients_count 2 100 10 -1
exec kill -SIGCONT $slave_pid
assert_equal [$rd read] {1}
assert_equal [$rd2 read] {1}
$rd ping
assert_equal [$rd read] {PONG}
$rd2 ping
assert_equal [$rd2 read] {PONG}
$rd close
$rd2 close
}
}} }}
...@@ -101,13 +131,13 @@ tags {"wait aof network external:skip"} { ...@@ -101,13 +131,13 @@ tags {"wait aof network external:skip"} {
$rd waitaof 1 0 0 $rd waitaof 1 0 0
wait_for_blocked_client wait_for_blocked_client
r config set appendonly no ;# this should release the blocked client as an error r config set appendonly no ;# this should release the blocked client as an error
assert_error {ERR WAITAOF cannot be used when appendonly is disabled} {$rd read} assert_error {ERR WAITAOF cannot be used when numlocal is set but appendonly is disabled.} {$rd read}
$rd close $rd close
} }
test {WAITAOF local on server with aof disabled} { test {WAITAOF local on server with aof disabled} {
$master incr foo $master incr foo
assert_error {ERR WAITAOF cannot be used when appendonly is disabled} {$master waitaof 1 0 0} assert_error {ERR WAITAOF cannot be used when numlocal is set but appendonly is disabled.} {$master waitaof 1 0 0}
} }
$master config set appendonly yes $master config set appendonly yes
...@@ -164,6 +194,36 @@ tags {"wait aof network external:skip"} { ...@@ -164,6 +194,36 @@ tags {"wait aof network external:skip"} {
assert_equal [$master waitaof 0 1 0] {1 1} assert_equal [$master waitaof 0 1 0] {1 1}
} }
test {WAITAOF replica multiple clients unblock - reuse last result} {
set rd [redis_deferring_client -1]
set rd2 [redis_deferring_client -1]
exec kill -SIGSTOP $replica_pid
$rd incr foo
$rd read
$rd2 incr foo
$rd2 read
$rd waitaof 0 1 0
$rd2 waitaof 0 1 0
wait_for_blocked_clients_count 2 100 10 -1
exec kill -SIGCONT $replica_pid
assert_equal [$rd read] {1 1}
assert_equal [$rd2 read] {1 1}
$rd ping
assert_equal [$rd read] {PONG}
$rd2 ping
assert_equal [$rd2 read] {PONG}
$rd close
$rd2 close
}
test {WAITAOF on promoted replica} { test {WAITAOF on promoted replica} {
$replica replicaof no one $replica replicaof no one
$replica incr foo $replica incr foo
......
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