
This PR is 1 of 3 PRs intended to achieve the goal of 1 million requests
per second, as detailed by [dan touitou](https://github.com/touitou-dan)
in https://github.com/valkey-io/valkey/issues/22. This PR modifies the
IO threads to be fully asynchronous, which is a first and necessary step
to allow more work offloading and better utilization of the IO threads.
### Current IO threads state:
Valkey IO threads were introduced in Redis 6.0 to allow better
utilization of multi-core machines. Before this, Redis was
single-threaded and could only use one CPU core for network and command
processing. The introduction of IO threads helps in offloading the IO
operations to multiple threads.
**Current IO Threads flow:**
1. Initialization: When Redis starts, it initializes a specified number
of IO threads. These threads are in addition to the main thread, each
thread starts with an empty list, the main thread will populate that
list in each event-loop with pending-read-clients or
pending-write-clients.
2. Read Phase: The main thread accepts incoming connections and reads
requests from clients. The reading of requests are offloaded to IO
threads. The main thread puts the clients ready-to-read in a list and
set the global io_threads_op to IO_THREADS_OP_READ, the IO threads pick
the clients up, perform the read operation and parse the first incoming
command.
3. Command Processing: After reading the requests, command processing is
still single-threaded and handled by the main thread.
4. Write Phase: Similar to the read phase, the write phase is also be
offloaded to IO threads. The main thread prepares the response in the
clients’ output buffer then the main thread puts the client in the list,
and sets the global io_threads_op to the IO_THREADS_OP_WRITE. The IO
threads then pick the clients up and perform the write operation to send
the responses back to clients.
5. Synchronization: The main-thread communicate with the threads on how
many jobs left per each thread with atomic counter. The main-thread
doesn’t access the clients while being handled by the IO threads.
**Issues with current implementation:**
* Underutilized Cores: The current implementation of IO-threads leads to
the underutilization of CPU cores.
* The main thread remains responsible for a significant portion of
IO-related tasks that could be offloaded to IO-threads.
* When the main-thread is processing client’s commands, the IO threads
are idle for a considerable amount of time.
* Notably, the main thread's performance during the IO-related tasks is
constrained by the speed of the slowest IO-thread.
* Limited Offloading: Currently, Since the Main-threads waits
synchronously for the IO threads, the Threads perform only read-parse,
and write operations, with parsing done only for the first command. If
the threads can do work asynchronously we may offload more work to the
threads reducing the load from the main-thread.
* TLS: Currently, we don't support IO threads with TLS (where offloading
IO would be more beneficial) since TLS read/write operations are not
thread-safe with the current implementation.
### Suggested change
Non-blocking main thread - The main thread and IO threads will operate
in parallel to maximize efficiency. The main thread will not be blocked
by IO operations. It will continue to process commands independently of
the IO thread's activities.
**Implementation details**
**Inter-thread communication.**
* We use a static, lock-free ring buffer of fixed size (2048 jobs) for
the main thread to send jobs and for the IO to receive them. If the ring
buffer fills up, the main thread will handle the task itself, acting as
back pressure (in case IO operations are more expensive than command
processing). A static ring buffer is a better candidate than a dynamic
job queue as it eliminates the need for allocation/freeing per job.
* An IO job will be in the format: ` [void* function-call-back | void
*data] `where data is either a client to read/write from and the
function-ptr is the function to be called with the data for example
readQueryFromClient using this format we can use it later to offload
other types of works to the IO threads.
* The Ring buffer is one way from the main-thread to the IO thread, Upon
read/write event the main thread will send a read/write job then in
before sleep it will iterate over the pending read/write clients to
checking for each client if the IO threads has already finished handling
it. The IO thread signals it has finished handling a client read/write
by toggling an atomic flag read_state / write_state on the client
struct.
**Thread Safety**
As suggested in this solution, the IO threads are reading from and
writing to the clients' buffers while the main thread may access those
clients.
We must ensure no race conditions or unsafe access occurs while keeping
the Valkey code simple and lock free.
Minimal Action in the IO Threads
The main change is to limit the IO thread operations to the bare
minimum. The IO thread will access only the client's struct and only the
necessary fields in this struct.
The IO threads will be responsible for the following:
* Read Operation: The IO thread will only read and parse a single
command. It will not update the server stats, handle read errors, or
parsing errors. These tasks will be taken care of by the main thread.
* Write Operation: The IO thread will only write the available data. It
will not free the client's replies, handle write errors, or update the
server statistics.
To achieve this without code duplication, the read/write code has been
refactored into smaller, independent components:
* Functions that perform only the read/parse/write calls.
* Functions that handle the read/parse/write results.
This refactor accounts for the majority of the modifications in this PR.
**Client Struct Safe Access**
As we ensure that the IO threads access memory only within the client
struct, we need to ensure thread safety only for the client's struct's
shared fields.
* Query Buffer
* Command parsing - The main thread will not try to parse a command from
the query buffer when a client is offloaded to the IO thread.
* Client's memory checks in client-cron - The main thread will not
access the client query buffer if it is offloaded and will handle the
querybuf grow/shrink when the client is back.
* CLIENT LIST command - The main thread will busy-wait for the IO thread
to finish handling the client, falling back to the current behavior
where the main thread waits for the IO thread to finish their
processing.
* Output Buffer
* The IO thread will not change the client's bufpos and won't free the
client's reply lists. These actions will be done by the main thread on
the client's return from the IO thread.
* bufpos / block→used: As the main thread may change the bufpos, the
reply-block→used, or add/delete blocks to the reply list while the IO
thread writes, we add two fields to the client struct: io_last_bufpos
and io_last_reply_block. The IO thread will write until the
io_last_bufpos, which was set by the main-thread before sending the
client to the IO thread. If more data has been added to the cob in
between, it will be written in the next write-job. In addition, the main
thread will not trim or merge reply blocks while the client is
offloaded.
* Parsing Fields
* Client's cmd, argc, argv, reqtype, etc., are set during parsing.
* The main thread will indicate to the IO thread not to parse a cmd if
the client is not reset. In this case, the IO thread will only read from
the network and won't attempt to parse a new command.
* The main thread won't access the c→cmd/c→argv in the CLIENT LIST
command as stated before it will busy wait for the IO threads.
* Client Flags
* c→flags, which may be changed by the main thread in multiple places,
won't be accessed by the IO thread. Instead, the main thread will set
the c→io_flags with the information necessary for the IO thread to know
the client's state.
* Client Close
* On freeClient, the main thread will busy wait for the IO thread to
finish processing the client's read/write before proceeding to free the
client.
* Client's Memory Limits
* The IO thread won't handle the qb/cob limits. In case a client crosses
the qb limit, the IO thread will stop reading for it, letting the main
thread know that the client crossed the limit.
**TLS**
TLS is currently not supported with IO threads for the following
reasons:
1. Pending reads - If SSL has pending data that has already been read
from the socket, there is a risk of not calling the read handler again.
To handle this, a list is used to hold the pending clients. With IO
threads, multiple threads can access the list concurrently.
2. Event loop modification - Currently, the TLS code
registers/unregisters the file descriptor from the event loop depending
on the read/write results. With IO threads, multiple threads can modify
the event loop struct simultaneously.
3. The same client can be sent to 2 different threads concurrently
(https://github.com/redis/redis/issues/12540).
Those issues were handled in the current PR:
1. The IO thread only performs the read operation. The main thread will
check for pending reads after the client returns from the IO thread and
will be the only one to access the pending list.
2. The registering/unregistering of events will be similarly postponed
and handled by the main thread only.
3. Each client is being sent to the same dedicated thread (c→id %
num_of_threads).
**Sending Replies Immediately with IO threads.**
Currently, after processing a command, we add the client to the
pending_writes_list. Only after processing all the clients do we send
all the replies. Since the IO threads are now working asynchronously, we
can send the reply immediately after processing the client’s requests,
reducing the command latency. However, if we are using AOF=always, we
must wait for the AOF buffer to be written, in which case we revert to
the current behavior.
**IO threads dynamic adjustment**
Currently, we use an all-or-nothing approach when activating the IO
threads. The current logic is as follows: if the number of pending write
clients is greater than twice the number of threads (including the main
thread), we enable all threads; otherwise, we enable none. For example,
if 8 IO threads are defined, we enable all 8 threads if there are 16
pending clients; else, we enable none.
It makes more sense to enable partial activation of the IO threads. If
we have 10 pending clients, we will enable 5 threads, and so on. This
approach allows for a more granular and efficient allocation of
resources based on the current workload.
In addition, the user will now be able to change the number of I/O
threads at runtime. For example, when decreasing the number of threads
from 4 to 2, threads 3 and 4 will be closed after flushing their job
queues.
**Tests**
Currently, we run the io-threads tests with 4 IO threads
(443d80f168/.github/workflows/daily.yml (L353)
).
This means that we will not activate the IO threads unless there are 8
(threads * 2) pending write clients per single loop, which is unlikely
to happened in most of tests, meaning the IO threads are not currently
being tested.
To enforce the main thread to always offload work to the IO threads,
regardless of the number of pending events, we add an
events-per-io-thread configuration with a default value of 2. When set
to 0, this configuration will force the main thread to always offload
work to the IO threads.
When we offload every single read/write operation to the IO threads, the
IO-threads are running with 100% CPU when running multiple tests
concurrently some tests fail as a result of larger than expected command
latencies. To address this issue, we have to add some after or wait_for
calls to some of the tests to ensure they pass with IO threads as well.
Signed-off-by: Uri Yagelnik <uriy@amazon.com>
612 lines
22 KiB
Tcl
612 lines
22 KiB
Tcl
start_server {tags {"maxmemory" "external:skip"}} {
|
|
r config set maxmemory 11mb
|
|
r config set maxmemory-policy allkeys-lru
|
|
set server_pid [s process_id]
|
|
|
|
proc init_test {client_eviction} {
|
|
r flushdb
|
|
|
|
set prev_maxmemory_clients [r config get maxmemory-clients]
|
|
if $client_eviction {
|
|
r config set maxmemory-clients 3mb
|
|
r client no-evict on
|
|
} else {
|
|
r config set maxmemory-clients 0
|
|
}
|
|
|
|
r config resetstat
|
|
# fill 5mb using 50 keys of 100kb
|
|
for {set j 0} {$j < 50} {incr j} {
|
|
r setrange $j 100000 x
|
|
}
|
|
assert_equal [r dbsize] 50
|
|
}
|
|
|
|
# Return true if the eviction occurred (client or key) based on argument
|
|
proc check_eviction_test {client_eviction} {
|
|
set evicted_keys [s evicted_keys]
|
|
set evicted_clients [s evicted_clients]
|
|
set dbsize [r dbsize]
|
|
|
|
if $client_eviction {
|
|
return [expr $evicted_clients > 0 && $evicted_keys == 0 && $dbsize == 50]
|
|
} else {
|
|
return [expr $evicted_clients == 0 && $evicted_keys > 0 && $dbsize < 50]
|
|
}
|
|
}
|
|
|
|
# Assert the eviction test passed (and prints some debug info on verbose)
|
|
proc verify_eviction_test {client_eviction} {
|
|
set evicted_keys [s evicted_keys]
|
|
set evicted_clients [s evicted_clients]
|
|
set dbsize [r dbsize]
|
|
|
|
if $::verbose {
|
|
puts "evicted keys: $evicted_keys"
|
|
puts "evicted clients: $evicted_clients"
|
|
puts "dbsize: $dbsize"
|
|
}
|
|
|
|
assert [check_eviction_test $client_eviction]
|
|
}
|
|
|
|
foreach {client_eviction} {false true} {
|
|
set clients {}
|
|
test "eviction due to output buffers of many MGET clients, client eviction: $client_eviction" {
|
|
init_test $client_eviction
|
|
|
|
for {set j 0} {$j < 20} {incr j} {
|
|
set rr [valkey_deferring_client]
|
|
lappend clients $rr
|
|
}
|
|
|
|
# Generate client output buffers via MGET until we can observe some effect on
|
|
# keys / client eviction, or we time out.
|
|
set t [clock seconds]
|
|
while {![check_eviction_test $client_eviction] && [expr [clock seconds] - $t] < 20} {
|
|
foreach rr $clients {
|
|
if {[catch {
|
|
$rr mget 1
|
|
$rr flush
|
|
} err]} {
|
|
lremove clients $rr
|
|
}
|
|
}
|
|
}
|
|
|
|
verify_eviction_test $client_eviction
|
|
}
|
|
foreach rr $clients {
|
|
$rr close
|
|
}
|
|
|
|
set clients {}
|
|
test "eviction due to input buffer of a dead client, client eviction: $client_eviction" {
|
|
init_test $client_eviction
|
|
|
|
for {set j 0} {$j < 30} {incr j} {
|
|
set rr [valkey_deferring_client]
|
|
lappend clients $rr
|
|
}
|
|
|
|
foreach rr $clients {
|
|
if {[catch {
|
|
$rr write "*250\r\n"
|
|
for {set j 0} {$j < 249} {incr j} {
|
|
$rr write "\$1000\r\n"
|
|
$rr write [string repeat x 1000]
|
|
$rr write "\r\n"
|
|
$rr flush
|
|
}
|
|
after 100; # give the server some time to process the input buffer - this was added to make sure the test pass with io-threads active.
|
|
}]} {
|
|
lremove clients $rr
|
|
}
|
|
}
|
|
|
|
verify_eviction_test $client_eviction
|
|
}
|
|
foreach rr $clients {
|
|
$rr close
|
|
}
|
|
|
|
set clients {}
|
|
test "eviction due to output buffers of pubsub, client eviction: $client_eviction" {
|
|
init_test $client_eviction
|
|
|
|
for {set j 0} {$j < 20} {incr j} {
|
|
set rr [valkey_client]
|
|
lappend clients $rr
|
|
}
|
|
|
|
foreach rr $clients {
|
|
$rr subscribe bla
|
|
}
|
|
|
|
# Generate client output buffers via PUBLISH until we can observe some effect on
|
|
# keys / client eviction, or we time out.
|
|
set bigstr [string repeat x 100000]
|
|
set t [clock seconds]
|
|
while {![check_eviction_test $client_eviction] && [expr [clock seconds] - $t] < 20} {
|
|
if {[catch { r publish bla $bigstr } err]} {
|
|
if $::verbose {
|
|
puts "Error publishing: $err"
|
|
}
|
|
}
|
|
}
|
|
|
|
verify_eviction_test $client_eviction
|
|
}
|
|
foreach rr $clients {
|
|
$rr close
|
|
}
|
|
}
|
|
|
|
}
|
|
|
|
start_server {tags {"maxmemory external:skip"}} {
|
|
test "Without maxmemory small integers are shared" {
|
|
r config set maxmemory 0
|
|
r set a 1
|
|
assert_refcount_morethan a 1
|
|
}
|
|
|
|
test "With maxmemory and non-LRU policy integers are still shared" {
|
|
r config set maxmemory 1073741824
|
|
r config set maxmemory-policy allkeys-random
|
|
r set a 1
|
|
assert_refcount_morethan a 1
|
|
}
|
|
|
|
test "With maxmemory and LRU policy integers are not shared" {
|
|
r config set maxmemory 1073741824
|
|
r config set maxmemory-policy allkeys-lru
|
|
r set a 1
|
|
r config set maxmemory-policy volatile-lru
|
|
r set b 1
|
|
assert_refcount 1 a
|
|
assert_refcount 1 b
|
|
r config set maxmemory 0
|
|
}
|
|
|
|
test "Shared integers are unshared with maxmemory and LRU policy" {
|
|
r set a 1
|
|
r set b 1
|
|
assert_refcount_morethan a 1
|
|
assert_refcount_morethan b 1
|
|
r config set maxmemory 1073741824
|
|
r config set maxmemory-policy allkeys-lru
|
|
r get a
|
|
assert_refcount 1 a
|
|
r config set maxmemory-policy volatile-lru
|
|
r get b
|
|
assert_refcount 1 b
|
|
r config set maxmemory 0
|
|
}
|
|
|
|
foreach policy {
|
|
allkeys-random allkeys-lru allkeys-lfu volatile-lru volatile-lfu volatile-random volatile-ttl
|
|
} {
|
|
test "maxmemory - is the memory limit honoured? (policy $policy)" {
|
|
# make sure to start with a blank instance
|
|
r flushall
|
|
# Get the current memory limit and calculate a new limit.
|
|
# We just add 100k to the current memory size so that it is
|
|
# fast for us to reach that limit.
|
|
set used [s used_memory]
|
|
set limit [expr {$used+100*1024}]
|
|
r config set maxmemory $limit
|
|
r config set maxmemory-policy $policy
|
|
# Now add keys until the limit is almost reached.
|
|
set numkeys 0
|
|
while 1 {
|
|
r setex [randomKey] 10000 x
|
|
incr numkeys
|
|
if {[s used_memory]+4096 > $limit} {
|
|
assert {$numkeys > 10}
|
|
break
|
|
}
|
|
}
|
|
# If we add the same number of keys already added again, we
|
|
# should still be under the limit.
|
|
for {set j 0} {$j < $numkeys} {incr j} {
|
|
r setex [randomKey] 10000 x
|
|
}
|
|
assert {[s used_memory] < ($limit+4096)}
|
|
}
|
|
}
|
|
|
|
foreach policy {
|
|
allkeys-random allkeys-lru volatile-lru volatile-random volatile-ttl
|
|
} {
|
|
test "maxmemory - only allkeys-* should remove non-volatile keys ($policy)" {
|
|
# make sure to start with a blank instance
|
|
r flushall
|
|
# Get the current memory limit and calculate a new limit.
|
|
# We just add 100k to the current memory size so that it is
|
|
# fast for us to reach that limit.
|
|
set used [s used_memory]
|
|
set limit [expr {$used+100*1024}]
|
|
r config set maxmemory $limit
|
|
r config set maxmemory-policy $policy
|
|
# Now add keys until the limit is almost reached.
|
|
set numkeys 0
|
|
while 1 {
|
|
r set [randomKey] x
|
|
incr numkeys
|
|
if {[s used_memory]+4096 > $limit} {
|
|
assert {$numkeys > 10}
|
|
break
|
|
}
|
|
}
|
|
# If we add the same number of keys already added again and
|
|
# the policy is allkeys-* we should still be under the limit.
|
|
# Otherwise we should see an error reported by Server.
|
|
set err 0
|
|
for {set j 0} {$j < $numkeys} {incr j} {
|
|
if {[catch {r set [randomKey] x} e]} {
|
|
if {[string match {*used memory*} $e]} {
|
|
set err 1
|
|
}
|
|
}
|
|
}
|
|
if {[string match allkeys-* $policy]} {
|
|
assert {[s used_memory] < ($limit+4096)}
|
|
} else {
|
|
assert {$err == 1}
|
|
}
|
|
}
|
|
}
|
|
|
|
foreach policy {
|
|
volatile-lru volatile-lfu volatile-random volatile-ttl
|
|
} {
|
|
test "maxmemory - policy $policy should only remove volatile keys." {
|
|
# make sure to start with a blank instance
|
|
r flushall
|
|
# Get the current memory limit and calculate a new limit.
|
|
# We just add 100k to the current memory size so that it is
|
|
# fast for us to reach that limit.
|
|
set used [s used_memory]
|
|
set limit [expr {$used+100*1024}]
|
|
r config set maxmemory $limit
|
|
r config set maxmemory-policy $policy
|
|
# Now add keys until the limit is almost reached.
|
|
set numkeys 0
|
|
while 1 {
|
|
# Odd keys are volatile
|
|
# Even keys are non volatile
|
|
if {$numkeys % 2} {
|
|
r setex "key:$numkeys" 10000 x
|
|
} else {
|
|
r set "key:$numkeys" x
|
|
}
|
|
if {[s used_memory]+4096 > $limit} {
|
|
assert {$numkeys > 10}
|
|
break
|
|
}
|
|
incr numkeys
|
|
}
|
|
# Now we add the same number of volatile keys already added.
|
|
# We expect the server to evict only volatile keys in order to make
|
|
# space.
|
|
set err 0
|
|
for {set j 0} {$j < $numkeys} {incr j} {
|
|
catch {r setex "foo:$j" 10000 x}
|
|
}
|
|
# We should still be under the limit.
|
|
assert {[s used_memory] < ($limit+4096)}
|
|
# However all our non volatile keys should be here.
|
|
for {set j 0} {$j < $numkeys} {incr j 2} {
|
|
assert {[r exists "key:$j"]}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
# Calculate query buffer memory of slave
|
|
proc slave_query_buffer {srv} {
|
|
set clients [split [$srv client list] "\r\n"]
|
|
set c [lsearch -inline $clients *flags=S*]
|
|
if {[string length $c] > 0} {
|
|
assert {[regexp {qbuf=([0-9]+)} $c - qbuf]}
|
|
assert {[regexp {qbuf-free=([0-9]+)} $c - qbuf_free]}
|
|
return [expr $qbuf + $qbuf_free]
|
|
}
|
|
return 0
|
|
}
|
|
|
|
proc test_slave_buffers {test_name cmd_count payload_len limit_memory pipeline} {
|
|
start_server {tags {"maxmemory external:skip"}} {
|
|
start_server {} {
|
|
set slave_pid [s process_id]
|
|
test "$test_name" {
|
|
set slave [srv 0 client]
|
|
set slave_host [srv 0 host]
|
|
set slave_port [srv 0 port]
|
|
set master [srv -1 client]
|
|
set master_host [srv -1 host]
|
|
set master_port [srv -1 port]
|
|
|
|
# Disable slow log for master to avoid memory growth in slow env.
|
|
$master config set slowlog-log-slower-than -1
|
|
|
|
# add 100 keys of 100k (10MB total)
|
|
for {set j 0} {$j < 100} {incr j} {
|
|
$master setrange "key:$j" 100000 asdf
|
|
}
|
|
|
|
# make sure master doesn't disconnect slave because of timeout
|
|
$master config set repl-timeout 1200 ;# 20 minutes (for valgrind and slow machines)
|
|
$master config set maxmemory-policy allkeys-random
|
|
$master config set client-output-buffer-limit "replica 100000000 100000000 300"
|
|
$master config set repl-backlog-size [expr {10*1024}]
|
|
|
|
# disable latency tracking
|
|
$master config set latency-tracking no
|
|
$slave config set latency-tracking no
|
|
|
|
$slave slaveof $master_host $master_port
|
|
wait_for_condition 50 100 {
|
|
[s 0 master_link_status] eq {up}
|
|
} else {
|
|
fail "Replication not started."
|
|
}
|
|
|
|
# measure used memory after the slave connected and set maxmemory
|
|
set orig_used [s -1 used_memory]
|
|
set orig_client_buf [s -1 mem_clients_normal]
|
|
set orig_mem_not_counted_for_evict [s -1 mem_not_counted_for_evict]
|
|
set orig_used_no_repl [expr {$orig_used - $orig_mem_not_counted_for_evict}]
|
|
set limit [expr {$orig_used - $orig_mem_not_counted_for_evict + 32*1024}]
|
|
|
|
if {$limit_memory==1} {
|
|
$master config set maxmemory $limit
|
|
}
|
|
|
|
# put the slave to sleep
|
|
set rd_slave [valkey_deferring_client]
|
|
pause_process $slave_pid
|
|
|
|
# send some 10mb worth of commands that don't increase the memory usage
|
|
if {$pipeline == 1} {
|
|
set rd_master [valkey_deferring_client -1]
|
|
for {set k 0} {$k < $cmd_count} {incr k} {
|
|
$rd_master setrange key:0 0 [string repeat A $payload_len]
|
|
}
|
|
for {set k 0} {$k < $cmd_count} {incr k} {
|
|
$rd_master read
|
|
}
|
|
} else {
|
|
for {set k 0} {$k < $cmd_count} {incr k} {
|
|
$master setrange key:0 0 [string repeat A $payload_len]
|
|
}
|
|
}
|
|
|
|
set new_used [s -1 used_memory]
|
|
set slave_buf [s -1 mem_clients_slaves]
|
|
set client_buf [s -1 mem_clients_normal]
|
|
set mem_not_counted_for_evict [s -1 mem_not_counted_for_evict]
|
|
set used_no_repl [expr {$new_used - $mem_not_counted_for_evict - [slave_query_buffer $master]}]
|
|
# we need to exclude replies buffer and query buffer of replica from used memory.
|
|
# removing the replica (output) buffers is done so that we are able to measure any other
|
|
# changes to the used memory and see that they're insignificant (the test's purpose is to check that
|
|
# the replica buffers are counted correctly, so the used memory growth after deducting them
|
|
# should be nearly 0).
|
|
# we remove the query buffers because on slow test platforms, they can accumulate many ACKs.
|
|
set delta [expr {($used_no_repl - $client_buf) - ($orig_used_no_repl - $orig_client_buf)}]
|
|
|
|
assert {[$master dbsize] == 100}
|
|
assert {$slave_buf > 2*1024*1024} ;# some of the data may have been pushed to the OS buffers
|
|
set delta_max [expr {$cmd_count / 2}] ;# 1 byte unaccounted for, with 1M commands will consume some 1MB
|
|
assert {$delta < $delta_max && $delta > -$delta_max}
|
|
|
|
$master client kill type slave
|
|
set info_str [$master info memory]
|
|
set killed_used [getInfoProperty $info_str used_memory]
|
|
set killed_mem_not_counted_for_evict [getInfoProperty $info_str mem_not_counted_for_evict]
|
|
set killed_slave_buf [s -1 mem_clients_slaves]
|
|
# we need to exclude replies buffer and query buffer of slave from used memory after kill slave
|
|
set killed_used_no_repl [expr {$killed_used - $killed_mem_not_counted_for_evict - [slave_query_buffer $master]}]
|
|
set delta_no_repl [expr {$killed_used_no_repl - $used_no_repl}]
|
|
assert {[$master dbsize] == 100}
|
|
assert {$killed_slave_buf == 0}
|
|
assert {$delta_no_repl > -$delta_max && $delta_no_repl < $delta_max}
|
|
|
|
}
|
|
# unfreeze slave process (after the 'test' succeeded or failed, but before we attempt to terminate the server
|
|
resume_process $slave_pid
|
|
}
|
|
}
|
|
}
|
|
|
|
# test that slave buffer are counted correctly
|
|
# we wanna use many small commands, and we don't wanna wait long
|
|
# so we need to use a pipeline (valkey_deferring_client)
|
|
# that may cause query buffer to fill and induce eviction, so we disable it
|
|
test_slave_buffers {slave buffer are counted correctly} 1000000 10 0 1
|
|
|
|
# test that slave buffer don't induce eviction
|
|
# test again with fewer (and bigger) commands without pipeline, but with eviction
|
|
test_slave_buffers "replica buffer don't induce eviction" 100000 100 1 0
|
|
|
|
start_server {tags {"maxmemory external:skip"}} {
|
|
test {Don't rehash if used memory exceeds maxmemory after rehash} {
|
|
r config set latency-tracking no
|
|
r config set maxmemory 0
|
|
r config set maxmemory-policy allkeys-random
|
|
|
|
# Next rehash size is 8192, that will eat 64k memory
|
|
populate 4095 "" 1
|
|
|
|
set used [s used_memory]
|
|
set limit [expr {$used + 10*1024}]
|
|
r config set maxmemory $limit
|
|
|
|
# Adding a key to meet the 1:1 radio.
|
|
r set k0 v0
|
|
# The dict has reached 4096, it can be resized in tryResizeHashTables in cron,
|
|
# or we add a key to let it check whether it can be resized.
|
|
r set k1 v1
|
|
# Next writing command will trigger evicting some keys if last
|
|
# command trigger DB dict rehash
|
|
r set k2 v2
|
|
# There must be 4098 keys because the server doesn't evict keys.
|
|
r dbsize
|
|
} {4098}
|
|
}
|
|
|
|
start_server {tags {"maxmemory external:skip"}} {
|
|
test {client tracking don't cause eviction feedback loop} {
|
|
r config set latency-tracking no
|
|
r config set maxmemory 0
|
|
r config set maxmemory-policy allkeys-lru
|
|
r config set maxmemory-eviction-tenacity 100
|
|
|
|
# 10 clients listening on tracking messages
|
|
set clients {}
|
|
for {set j 0} {$j < 10} {incr j} {
|
|
lappend clients [valkey_deferring_client]
|
|
}
|
|
foreach rd $clients {
|
|
$rd HELLO 3
|
|
$rd read ; # Consume the HELLO reply
|
|
$rd CLIENT TRACKING on
|
|
$rd read ; # Consume the CLIENT reply
|
|
}
|
|
|
|
# populate 300 keys, with long key name and short value
|
|
for {set j 0} {$j < 300} {incr j} {
|
|
set key $j[string repeat x 1000]
|
|
r set $key x
|
|
|
|
# for each key, enable caching for this key
|
|
foreach rd $clients {
|
|
$rd get $key
|
|
$rd read
|
|
}
|
|
}
|
|
|
|
# we need to wait one second for the client querybuf excess memory to be
|
|
# trimmed by cron, otherwise the INFO used_memory and CONFIG maxmemory
|
|
# below (on slow machines) won't be "atomic" and won't trigger eviction.
|
|
after 1100
|
|
|
|
# set the memory limit which will cause a few keys to be evicted
|
|
# we need to make sure to evict keynames of a total size of more than
|
|
# 16kb since the (PROTO_REPLY_CHUNK_BYTES), only after that the
|
|
# invalidation messages have a chance to trigger further eviction.
|
|
set used [s used_memory]
|
|
set limit [expr {$used - 40000}]
|
|
r config set maxmemory $limit
|
|
|
|
# make sure some eviction happened
|
|
set evicted [s evicted_keys]
|
|
if {$::verbose} { puts "evicted: $evicted" }
|
|
|
|
# make sure we didn't drain the database
|
|
assert_range [r dbsize] 200 300
|
|
|
|
assert_range $evicted 10 50
|
|
foreach rd $clients {
|
|
$rd read ;# make sure we have some invalidation message waiting
|
|
$rd close
|
|
}
|
|
|
|
# eviction continues (known problem described in #8069)
|
|
# for now this test only make sures the eviction loop itself doesn't
|
|
# have feedback loop
|
|
set evicted [s evicted_keys]
|
|
if {$::verbose} { puts "evicted: $evicted" }
|
|
}
|
|
}
|
|
|
|
start_server {tags {"maxmemory" "external:skip"}} {
|
|
test {propagation with eviction} {
|
|
set repl [attach_to_replication_stream]
|
|
|
|
r set asdf1 1
|
|
r set asdf2 2
|
|
r set asdf3 3
|
|
|
|
r config set maxmemory-policy allkeys-lru
|
|
r config set maxmemory 1
|
|
|
|
wait_for_condition 5000 10 {
|
|
[r dbsize] eq 0
|
|
} else {
|
|
fail "Not all keys have been evicted"
|
|
}
|
|
|
|
r config set maxmemory 0
|
|
r config set maxmemory-policy noeviction
|
|
|
|
r set asdf4 4
|
|
|
|
assert_replication_stream $repl {
|
|
{select *}
|
|
{set asdf1 1}
|
|
{set asdf2 2}
|
|
{set asdf3 3}
|
|
{del asdf*}
|
|
{del asdf*}
|
|
{del asdf*}
|
|
{set asdf4 4}
|
|
}
|
|
close_replication_stream $repl
|
|
|
|
r config set maxmemory 0
|
|
r config set maxmemory-policy noeviction
|
|
}
|
|
}
|
|
|
|
start_server {tags {"maxmemory" "external:skip"}} {
|
|
test {propagation with eviction in MULTI} {
|
|
set repl [attach_to_replication_stream]
|
|
|
|
r config set maxmemory-policy allkeys-lru
|
|
|
|
r multi
|
|
r incr x
|
|
r config set maxmemory 1
|
|
r incr x
|
|
assert_equal [r exec] {1 OK 2}
|
|
|
|
wait_for_condition 5000 10 {
|
|
[r dbsize] eq 0
|
|
} else {
|
|
fail "Not all keys have been evicted"
|
|
}
|
|
|
|
assert_replication_stream $repl {
|
|
{multi}
|
|
{select *}
|
|
{incr x}
|
|
{incr x}
|
|
{exec}
|
|
{del x}
|
|
}
|
|
close_replication_stream $repl
|
|
|
|
r config set maxmemory 0
|
|
r config set maxmemory-policy noeviction
|
|
}
|
|
}
|
|
|
|
start_server {tags {"maxmemory" "external:skip"}} {
|
|
test {lru/lfu value of the key just added} {
|
|
r config set maxmemory-policy allkeys-lru
|
|
r set foo a
|
|
assert {[r object idletime foo] <= 2}
|
|
r del foo
|
|
r set foo 1
|
|
r get foo
|
|
assert {[r object idletime foo] <= 2}
|
|
|
|
r config set maxmemory-policy allkeys-lfu
|
|
r del foo
|
|
r set foo a
|
|
assert {[r object freq foo] == 5}
|
|
}
|
|
}
|