Skip to content

[fix] io thread for swap #393

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
warrick1016 merged 2 commits into from
Feb 9, 2026
Merged

[fix] io thread for swap #393

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
0858ddf
[fix] io thread for swap
warrick1016 Feb 9, 2026
eacd133
[debug] read_enable_async_io
warrick1016 Feb 9, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion src/config.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3535,7 +3535,7 @@ standardConfig static_configs[] = {
createBoolConfig("rocksdb.meta.enable_blob_files", NULL, MODIFIABLE_CONFIG, server.rocksdb_meta_enable_blob_files, 0, NULL, updateRocksdbMetaEnableBlobFiles),
createBoolConfig("rocksdb.data.enable_blob_garbage_collection", "rocksdb.enable_blob_garbage_collection", MODIFIABLE_CONFIG, server.rocksdb_data_enable_blob_garbage_collection, 1, NULL, updateRocksdbDataEnableBlobGarbageCollection),
createBoolConfig("rocksdb.meta.enable_blob_garbage_collection", NULL, MODIFIABLE_CONFIG, server.rocksdb_meta_enable_blob_garbage_collection, 1, NULL, updateRocksdbMetaEnableBlobGarbageCollection),
createBoolConfig("rocksdb.read_enable_async_io", NULL, IMMUTABLE_CONFIG, server.rocksdb_read_enable_async_io, 1, NULL, NULL),
createBoolConfig("rocksdb.read_enable_async_io", NULL, IMMUTABLE_CONFIG, server.rocksdb_read_enable_async_io, 0, NULL, NULL),
#endif
/* String Configs */
createStringConfig("aclfile", NULL, IMMUTABLE_CONFIG, ALLOW_EMPTY_STRING, server.acl_filename, "", NULL, NULL),
Expand Down
289 changes: 289 additions & 0 deletions tests/swap/unit/io_thread.tcl
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,289 @@
proc get_info_field {info field} {
set fl [string length $field]
append field :
foreach line [split $info "\n"] {
set line [string trim $line "\r\n "]
if {[string range $line 0 $fl] eq $field} {
return [string range $line [expr {$fl+1}] end]
}
}
return {}
}

proc get_kv_value {input key} {
foreach pair [split $input ","] {
if {[regexp {^\s*([^=]+)\s*=\s*(.+?)\s*$} $pair -> k v]} {
if {$k eq $key} {
return $v
}
}
}
return ""
}

start_server {overrides {}} {
r set k v


test "threads 1 => n and n => 1" {
for {set thread_size 2} {$thread_size < 5} {incr thread_size} {
assert_equal [get_info_field [r info threads] io_thread_scale_status] "none"
assert {[get_info_field [r info threads] io_thread_1 ] eq ""}
# set io-threads n
# when client size < thread size , thread scale up task finish
if {!$::external} {
set lines [count_log_lines 0]
}
assert_equal [get_kv_value [get_info_field [r info threads] io_thread_0 ] clients] 1
r config set io-threads $thread_size
if {!$::external} {
wait_for_condition 200 50 {
[expr {![catch {verify_log_message 0 "*IO threads scale-up end*" $lines}]}]
} else {
fail "scale-up end log not found within timeout"
}
}
assert_equal [r get k] v


# reset io-threads 1
if {!$::external} {
set lines [count_log_lines 0]
}
r config set io-threads 1
after 200
assert_equal [get_info_field [r info threads] io_thread_scale_status] "down"
wait_for_condition 200 50 {
[get_info_field [r info threads] io_thread_1 ] eq ""
} else {
fail "thread down n => 1 fail"
}
if {!$::external} {
verify_log_message 0 "*IO threads scale-down end*" $lines
}

# add clients
set clients []
for {set j 0} {$j < 100} {incr j} {
set cli [redis [srv 0 "host"] [srv 0 "port"] 0 $::tls]
$cli select $::target_db
lappend clients $cli
}
after 200
wait_for_condition 200 50 {
[expr {[get_kv_value [get_info_field [r info threads] io_thread_0 ] clients] == 101}]
} else {
fail "io_thread_0 clients did not reach 101 within timeout"
}
# set io-threads n
if {!$::external} {
set lines [count_log_lines 0]
}
r config set io-threads $thread_size

if {!$::external} {
wait_for_condition 200 50 {
[expr {![catch {verify_log_message 0 "*IO threads scale-up end*" $lines}]}]
} else {
fail "scale-up end log not found within timeout"
}
}
assert_equal [r get k] v
for {set j 0} {$j < 100} {incr j} {
set cli [lindex $clients $j]
assert_equal [$cli get k] v
}


# reset io-threads 1
if {!$::external} {
set lines [count_log_lines 0]
}
r config set io-threads 1
assert_equal [get_info_field [r info threads] io_thread_scale_status] "down"
wait_for_condition 200 50 {
[get_info_field [r info threads] io_thread_1 ] eq ""
} else {
fail "thread down n => 1 fail"
}
if {!$::external} {
verify_log_message 0 "*IO threads scale-down end*" $lines
}

# close all clients
for {set j 0} {$j < 100} {incr j} {
set cli [lindex $clients $j]
$cli close
}
}


}

# test before set io-threads 2
r config set io-threads 2

test "threads 2 => n and n => 2" {
for {set thread_size 3} {$thread_size < 5} {incr thread_size} {
assert_equal [get_info_field [r info threads] io_thread_scale_status] "none"
assert {[get_info_field [r info threads] io_thread_1 ] ne ""}
assert {[get_info_field [r info threads] io_thread_2 ] eq ""}

# set io-threads n
# when client size < thread size , thread scale up task finish
if {!$::external} {
set lines [count_log_lines 0]
}
assert_equal [get_kv_value [get_info_field [r info threads] io_thread_1 ] clients] 1
r config set io-threads $thread_size
if {!$::external} {
wait_for_condition 200 50 {
[expr {![catch {verify_log_message 0 "*IO threads scale-up end*" $lines}]}]
} else {
fail "scale-up end log not found within timeout"
}
}


# reset io-threads 2
if {!$::external} {
set lines [count_log_lines 0]
}
r config set io-threads 2
assert_equal [get_info_field [r info threads] io_thread_scale_status] "down"
wait_for_condition 200 50 {
[get_info_field [r info threads] io_thread_2 ] eq ""
} else {
fail "thread down n => 2 fail"
}
if {!$::external} {
verify_log_message 0 "*IO threads scale-down end*" $lines
}

# add clients
set clients []
for {set j 0} {$j < 100} {incr j} {
set cli [redis [srv 0 "host"] [srv 0 "port"] 0 $::tls]
if {!$::singledb} {
$cli select $::target_db
}
lappend clients $cli
}
wait_for_condition 200 50 {
[expr {[get_kv_value [get_info_field [r info threads] io_thread_1 ] clients] == 101}]
} else {
fail "io_thread_1 clients did not reach 101 within timeout"
}

# set io-threads n
# wait CLIENT_IO_PENDING_CRON ,load balancing
if {!$::external} {
set lines [count_log_lines 0]
}
r config set io-threads $thread_size
assert_equal [get_info_field [r info threads] io_thread_scale_status] "up"
wait_for_condition 200 50 {
[get_info_field [r info threads] io_thread_scale_status] eq "none"
} else {
fail "thread up 2=>n fail"
}

if {!$::external} {
assert {[catch {verify_log_message 0 "*IO threads scale-up client num(1)< thread num*" $lines} errorMsg]}
assert {$errorMsg ne ""}
wait_for_condition 200 50 {
[expr {![catch {verify_log_message 0 "*IO threads scale-up end*" $lines}]}]
} else {
fail "scale-up end log not found within timeout"
}
}

# reset io-threads 2
# wait CLIENT_IO_PENDING_CRON
if {!$::external} {
set lines [count_log_lines 0]
}
r config set io-threads 2
assert_equal [get_info_field [r info threads] io_thread_scale_status] "down"
wait_for_condition 200 50 {
[get_info_field [r info threads] io_thread_2 ] eq ""
} else {
fail "thread down n => 2 fail"
}
wait_for_condition 200 50 {
[get_info_field [r info threads] io_thread_scale_status] eq "none"
} else {
fail "thread down n => 2 scale_status did not reach none"
}
if {!$::external} {
verify_log_message 0 "*IO threads scale-down end*" $lines
}

# set io-threads n
# client write, load balancing
if {!$::external} {
set lines [count_log_lines 0]
}
r config set io-threads $thread_size
assert_equal [get_info_field [r info threads] io_thread_scale_status] "up"
for {set j 0} {$j < 100} {incr j} {
set cli [lindex $clients $j]
assert_equal [$cli get k] v
}
assert_equal [get_info_field [r info threads] io_thread_scale_status] "none"
if {!$::external} {
assert {[catch {verify_log_message 0 "*IO threads scale-up client num(1)< thread num*" $lines} errorMsg]}
assert {$errorMsg ne ""}
wait_for_condition 200 50 {
[expr {![catch {verify_log_message 0 "*IO threads scale-up end*" $lines}]}]
} else {
fail "scale-up end log not found within timeout"
}
}


# reset io-threads 2
# client write
if {!$::external} {
set lines [count_log_lines 0]
}
r config set io-threads 2
assert_equal [get_info_field [r info threads] io_thread_scale_status] "down"
assert_equal [r get k] v
for {set j 0} {$j < 100} {incr j} {
set cli [lindex $clients $j]
assert_equal [$cli get k] v
}

set info [r info threads]
if {[get_info_field $info io_thread_2] ne ""} {
assert_equal [get_kv_value [get_info_field [r info threads] io_thread_2 ] clients] 0
# need wait thread_join
wait_for_condition 200 50 {
[get_info_field [r info threads] io_thread_scale_status] eq "none"
} else {
fail "thread down n => 2 fail"
}
} else {
assert_equal [get_info_field [r info threads] io_thread_scale_status] "none"
}


if {!$::external} {
verify_log_message 0 "*IO threads scale-down end*" $lines
}



# close all clients
for {set j 0} {$j < 100} {incr j} {
set cli [lindex $clients $j]
$cli close
}

}

}


}
2 changes: 1 addition & 1 deletion tests/unit/io_thread.tcl
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,7 +21,7 @@ proc get_kv_value {input key} {
return ""
}

start_server {overrides {}} {
start_server {tags {"memonly"}} {
r set k v


Expand Down
Loading