Commit 15c078df authored by yoav-steinberg's avatar yoav-steinberg Committed by Oran Agra
Browse files

Enforce client output buffer soft limit when no traffic. (#8833)



When client breached the output buffer soft limit but then went idle,
we didn't disconnect on soft limit timeout, now we do.
Note this also resolves some sporadic test failures in due to Linux
buffering data which caused tests to fail if during the test we went
back under the soft COB limit.
Co-authored-by: default avatarOran Agra <oran@redislabs.com>
Co-authored-by: default avatarsundb <sundbcn@gmail.com>
(cherry picked from commit 152fce5e)
parent e90e5640
...@@ -333,7 +333,7 @@ void _addReplyProtoToList(client *c, const char *s, size_t len) { ...@@ -333,7 +333,7 @@ void _addReplyProtoToList(client *c, const char *s, size_t len) {
listAddNodeTail(c->reply, tail); listAddNodeTail(c->reply, tail);
c->reply_bytes += tail->size; c->reply_bytes += tail->size;
asyncCloseClientOnOutputBufferLimitReached(c); closeClientOnOutputBufferLimitReached(c, 1);
} }
} }
...@@ -616,7 +616,7 @@ void setDeferredReply(client *c, void *node, const char *s, size_t length) { ...@@ -616,7 +616,7 @@ void setDeferredReply(client *c, void *node, const char *s, size_t length) {
listNodeValue(ln) = buf; listNodeValue(ln) = buf;
c->reply_bytes += buf->size; c->reply_bytes += buf->size;
asyncCloseClientOnOutputBufferLimitReached(c); closeClientOnOutputBufferLimitReached(c, 1);
} }
} }
...@@ -949,7 +949,7 @@ void AddReplyFromClient(client *dst, client *src) { ...@@ -949,7 +949,7 @@ void AddReplyFromClient(client *dst, client *src) {
src->bufpos = 0; src->bufpos = 0;
/* Check output buffer limits */ /* Check output buffer limits */
asyncCloseClientOnOutputBufferLimitReached(dst); closeClientOnOutputBufferLimitReached(dst, 1);
} }
/* Copy 'src' client output buffers into 'dst' client output buffers. /* Copy 'src' client output buffers into 'dst' client output buffers.
...@@ -3223,18 +3223,33 @@ int checkClientOutputBufferLimits(client *c) { ...@@ -3223,18 +3223,33 @@ int checkClientOutputBufferLimits(client *c) {
* *
* Note: we need to close the client asynchronously because this function is * Note: we need to close the client asynchronously because this function is
* called from contexts where the client can't be freed safely, i.e. from the * called from contexts where the client can't be freed safely, i.e. from the
* lower level functions pushing data inside the client output buffers. */ * lower level functions pushing data inside the client output buffers.
void asyncCloseClientOnOutputBufferLimitReached(client *c) { * When `async` is set to 0, we close the client immediately, this is
if (!c->conn) return; /* It is unsafe to free fake clients. */ * useful when called from cron.
*
* Returns 1 if client was (flagged) closed. */
int closeClientOnOutputBufferLimitReached(client *c, int async) {
if (!c->conn) return 0; /* It is unsafe to free fake clients. */
serverAssert(c->reply_bytes < SIZE_MAX-(1024*64)); serverAssert(c->reply_bytes < SIZE_MAX-(1024*64));
if (c->reply_bytes == 0 || c->flags & CLIENT_CLOSE_ASAP) return; if (c->reply_bytes == 0 || c->flags & CLIENT_CLOSE_ASAP) return 0;
if (checkClientOutputBufferLimits(c)) { if (checkClientOutputBufferLimits(c)) {
sds client = catClientInfoString(sdsempty(),c); sds client = catClientInfoString(sdsempty(),c);
freeClientAsync(c); if (async) {
serverLog(LL_WARNING,"Client %s scheduled to be closed ASAP for overcoming of output buffer limits.", client); freeClientAsync(c);
serverLog(LL_WARNING,
"Client %s scheduled to be closed ASAP for overcoming of output buffer limits.",
client);
} else {
freeClient(c);
serverLog(LL_WARNING,
"Client %s closed for overcoming of output buffer limits.",
client);
}
sdsfree(client); sdsfree(client);
return 1;
} }
return 0;
} }
/* Helper function used by performEvictions() in order to flush slaves /* Helper function used by performEvictions() in order to flush slaves
......
...@@ -1839,6 +1839,7 @@ void clientsCron(void) { ...@@ -1839,6 +1839,7 @@ void clientsCron(void) {
if (clientsCronResizeQueryBuffer(c)) continue; if (clientsCronResizeQueryBuffer(c)) continue;
if (clientsCronTrackExpansiveClients(c, curr_peak_mem_usage_slot)) continue; if (clientsCronTrackExpansiveClients(c, curr_peak_mem_usage_slot)) continue;
if (clientsCronTrackClientsMemUsage(c)) continue; if (clientsCronTrackClientsMemUsage(c)) continue;
if (closeClientOnOutputBufferLimitReached(c, 0)) continue;
} }
} }
......
...@@ -1867,7 +1867,7 @@ void rewriteClientCommandArgument(client *c, int i, robj *newval); ...@@ -1867,7 +1867,7 @@ void rewriteClientCommandArgument(client *c, int i, robj *newval);
void replaceClientCommandVector(client *c, int argc, robj **argv); void replaceClientCommandVector(client *c, int argc, robj **argv);
unsigned long getClientOutputBufferMemoryUsage(client *c); unsigned long getClientOutputBufferMemoryUsage(client *c);
int freeClientsInAsyncFreeQueue(void); int freeClientsInAsyncFreeQueue(void);
void asyncCloseClientOnOutputBufferLimitReached(client *c); int closeClientOnOutputBufferLimitReached(client *c, int async);
int getClientType(client *c); int getClientType(client *c);
int getClientTypeByName(char *name); int getClientTypeByName(char *name);
char *getClientTypeName(int class); char *getClientTypeName(int class);
......
...@@ -18,65 +18,72 @@ start_server {tags {"obuf-limits"}} { ...@@ -18,65 +18,72 @@ start_server {tags {"obuf-limits"}} {
assert {$omem >= 70000 && $omem < 200000} assert {$omem >= 70000 && $omem < 200000}
$rd1 close $rd1 close
} }
test {Client output buffer soft limit is not enforced if time is not overreached} { foreach {soft_limit_time wait_for_timeout} {3 yes
r config set client-output-buffer-limit {pubsub 0 100000 10} 4 no } {
set rd1 [redis_deferring_client] if $wait_for_timeout {
set test_name "Client output buffer soft limit is enforced if time is overreached"
$rd1 subscribe foo } else {
set reply [$rd1 read] set test_name "Client output buffer soft limit is not enforced too early and is enforced when no traffic"
assert {$reply eq "subscribe foo 1"}
set omem 0
set start_time 0
set time_elapsed 0
while 1 {
if {$start_time != 0} {
# Slow down loop when omen has reached the limit.
after 10
}
r publish foo [string repeat "x" 1000]
set clients [split [r client list] "\r\n"]
set c [split [lindex $clients 1] " "]
if {![regexp {omem=([0-9]+)} $c - omem]} break
if {$omem > 100000} {
if {$start_time == 0} {set start_time [clock seconds]}
set time_elapsed [expr {[clock seconds]-$start_time}]
if {$time_elapsed >= 5} break
}
} }
assert {$omem >= 100000 && $time_elapsed >= 5 && $time_elapsed <= 10}
$rd1 close
}
test {Client output buffer soft limit is enforced if time is overreached} {
r config set client-output-buffer-limit {pubsub 0 100000 3}
set rd1 [redis_deferring_client]
$rd1 subscribe foo
set reply [$rd1 read]
assert {$reply eq "subscribe foo 1"}
set omem 0 test $test_name {
set start_time 0 r config set client-output-buffer-limit "pubsub 0 100000 $soft_limit_time"
set time_elapsed 0 set soft_limit_time [expr $soft_limit_time*1000]
while 1 { set rd1 [redis_deferring_client]
if {$start_time != 0} {
# Slow down loop when omen has reached the limit. $rd1 client setname test_client
after 10 set reply [$rd1 read]
assert {$reply eq "OK"}
$rd1 subscribe foo
set reply [$rd1 read]
assert {$reply eq "subscribe foo 1"}
set omem 0
set start_time 0
set time_elapsed 0
set last_under_limit_time [clock milliseconds]
while 1 {
r publish foo [string repeat "x" 1000]
set clients [split [r client list] "\r\n"]
set c [lsearch -inline $clients *name=test_client*]
if {$start_time != 0} {
set time_elapsed [expr {[clock milliseconds]-$start_time}]
# Make sure test isn't taking too long
assert {$time_elapsed <= [expr $soft_limit_time+3000]}
}
if {$wait_for_timeout && $c == ""} {
# Make sure we're disconnected when we reach the soft limit
assert {$omem >= 100000 && $time_elapsed >= $soft_limit_time}
break
} else {
assert {[regexp {omem=([0-9]+)} $c - omem]}
}
if {$omem > 100000} {
if {$start_time == 0} {set start_time $last_under_limit_time}
if {!$wait_for_timeout && $time_elapsed >= [expr $soft_limit_time-1000]} break
# Slow down loop when omem has reached the limit.
after 10
} else {
# if the OS socket buffers swallowed what we previously filled, reset the start timer.
set start_time 0
set last_under_limit_time [clock milliseconds]
}
} }
r publish foo [string repeat "x" 1000]
set clients [split [r client list] "\r\n"] if {!$wait_for_timeout} {
set c [split [lindex $clients 1] " "] # After we completely stopped the traffic, wait for soft limit to time out
if {![regexp {omem=([0-9]+)} $c - omem]} break set timeout [expr {$soft_limit_time+1500 - ([clock milliseconds]-$start_time)}]
if {$omem > 100000} { wait_for_condition [expr $timeout/10] 10 {
if {$start_time == 0} {set start_time [clock seconds]} [lsearch [split [r client list] "\r\n"] *name=test_client*] == -1
set time_elapsed [expr {[clock seconds]-$start_time}] } else {
if {$time_elapsed >= 10} break fail "Soft limit timed out but client still connected"
}
} }
$rd1 close
} }
assert {$omem >= 100000 && $time_elapsed < 6}
$rd1 close
} }
test {No response for single command if client output buffer hard limit is enforced} { test {No response for single command if client output buffer hard limit is enforced} {
......
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