Table of Contents

If it’s broken, I’ll fix it!

Reintroducing an interesting problem (background)
#

Description: This write-up fixes RDMA crashes under multithreading. While we work on it, main is also refactoring the I/O model — we do not yet know if the new model still hits this. (Verify later.)

We also need to understand the new I/O model (study PR history later; see Feature: Redis IO model).

Roughly:

flowchart LR
    Main["主线程"]
    IO["IO 线程"]
    Inbox["io_shared_inbox<br/>SPMC"]
    Outbox["io_shared_outbox<br/>MPSC"]

    Main -->|"JOB_REQ_READ/WRITE"| Inbox
    Inbox --> IO
    IO --> Outbox
    Outbox -->|"JOB_RES_*"| Main

Main’s I/O model has changed, so first verify whether the old RDMA multithread bug still exists under the new model. I opened a branch to check — it still fails. Good.

Testing under the new I/O model
#

At the old intensity, failures are rare; e.g. the following often does not trip:

./src/valkey-benchmark -h 10.0.0.1 -p 6379 -d 256 --threads 16 -c 100 -P 32 -n 20000000 -t get --rdma

Reproduction shows it still exists, but rarely? → crank up the load.

# -c 500 增加并发，-P 256 极限流水线，极容易打满 RDMA 的网卡队列
./src/valkey-benchmark -h 10.0.0.1 -p 6379 -d 256 --threads 16 -c 500 -P 256 -n 50000000 -t set,get --rdma

This mixed workload is brutal — under the new I/O model it should hard-fail, still with the c->cmd_queue.len == 0' is not true assertion.

That command is so heavy that even our current code I/O-hangs immediately. Proof the bug is still there. (Is this load “fair”? Our parameters are unbounded — must every arbitrary stress command “work”? Either way, assertions must not fire. Slow is OK; segfaults are not.)

New I/O model: direct assertion failure
Old model: can also deadlock (or I/O hang? park that for now.)

# -d 8192 甚至更高，测试大块数据在 RDMA 传输时是否会导致内存踩踏或分配器崩溃
./src/valkey-benchmark -h 10.0.0.1 -p 6379 -d 8192 --threads 16 -c 200 -P 64 -n 10000000 -t set --rdma

This one is slow but eventually finishes — still smells wrong. Treat as OK for now. It no longer hangs forever — we are actually fixing things.

Keep trying to fix it
#

Build the server:

sudo prlimit --memlock=unlimited --pid $$
sudo modprobe dummy
sudo ip link add dummy0 type dummy
sudo ip link set dummy0 up
sudo ip addr add 10.0.0.1/24 dev dummy0
sudo rdma link add rxe_dummy type rxe netdev dummy0
sudo systemctl stop valkey
make -j$(nproc) BUILD_RDMA=yes USE_FAST_FLOAT=yes # 注意这里开启多线程的编译
sudo ./src/valkey-server valkey.conf --rdma-bind 10.0.0.1 --rdma-port 6379

Whenever the port is truly occupied:

sudo lsof -i :6379

lsof (List Open Files) lists files currently open on the system. On Linux “everything is a file”, including network sockets. -i :6379 filters connections on port 6379.

~/Project/valkey fix-rdma-assertion-new-io* ❯ sudo lsof -i :6379
COMMAND     PID USER FD   TYPE DEVICE SIZE/OFF NODE NAME
valkey-se 26355 root 8u  IPv4 148550      0t0  TCP *:redis (LISTEN)
~/Project/valkey fix-rdma-assertion-new-io* ❯ sudo kill 26355

Client test:

sudo prlimit --memlock=unlimited --pid $$
./src/valkey-benchmark -h 10.0.0.1 -p 6379 -d 256 --threads 16 -c 500 -P 256 -n 50000000 -t set,get --rdma

And our valkey.conf looks like this. Indeed single-thread is fine; only with multiple IO threads does it fail.

io-threads 4
save ""
protected-mode no
bind 0.0.0.0

And we can still reproduce the old error stably.

Detailed discussion
#

First deeply understand how IO moves client work — Feature: Redis IO model We must deeply understand how to fix this — our first deep technical discussion with pizhenwei. I drew an architecture diagram to pin the bug — proud of it (even if it looks ugly).

First declare / sort a few variables:

State machine	Meaning
`io_read_state` / `io_write_state`	offload lifecycle: `IDLE → PENDING_IO → COMPLETED_IO → IDLE`
`cmd_queue`	Pipeline commands already parsed by IO, not yet executed by main → that is what parsing feeds

The assertion means: do not parse new commands before prior ones have been executed.

RDMA quirk: no standalone POLLOUT; connUpdateState() → connRdmaEventHandler() sync-calls read_handler / write_handler.

While fixing, GET also hits an access-nil error
#

Multithread bugs are truly complex — too hard.

The error log is huge — put it in a file to analyze; most lines are noise, but then you lose the earliest error.

sudo ./src/valkey-server valkey.conf --rdma-bind 10.0.0.1 --rdma-port 6379 > valkey-run.log 2>&1

Maybe related to the old thread model in Multithread RDMA crash — keep digging. That engineer is sharp, but this looks like a different issue. Use:

sudo prlimit --memlock=unlimited --pid $$
./src/valkey-benchmark -h 10.0.0.1 -p 6379 -d 256 --threads 16 -c 500 -P 256 -n 50000000 -t set,get --rdma

Under load, entering the GET phase, Ctrl+C always crashes the server — another serious bug. How do you catch such a crash — whose call path blew up? First check whether the server binary has debug symbols:

~/Project/valkey fix-rdma-assertion-new-io* ❯ file ./src/valkey-server
./src/valkey-server: ELF 64-bit LSB pie executable, x86-64, version 1 (GNU/Linux), dynamically linked, interpreter /lib64/ld-linux-x86-64.so.2, BuildID[sha1]=afbcceb012f03c869136fe49f6f71261a65185af, for GNU/Linux 4.4.0, with debug_info, not stripped

Then inside gdb:

~/Project/valkey fix-rdma-assertion-new-io* ❯ sudo gdb --args ./src/valkey-server valkey.conf --rdma-bind 10.0.0.1 --rdma-port 6379

Start the server first:

set pagination off
handle SIGPIPE nostop noprint
run

After it crashes:

thread apply all bt
bt
frame 1
info frame

Inspect thread frames to see who called into the access-nil path. Very satisfying — you see exactly where it blew:

Thread 1 (Thread 0x7ffff7c90740 (LWP 42401) "valkey-server"):
#0  listUnlinkNode (list=0x7ffff743d420, node=0x0) at /home/ada/Project/valkey/src/adlist.c:194
#1  0x000055555570ef56 in listDelNode (list=0x7ffff743d420, node=0x0) at /home/ada/Project/valkey/src/adlist.c:183
#2  rdmaProcessPendingData () at /home/ada/Project/valkey/src/rdma.c:1785
#3  0x0000555555747f93 in connTypeProcessPendingData () at /home/ada/Project/valkey/src/connection.c:147
#4  beforeSleep (eventLoop=<optimized out>) at /home/ada/Project/valkey/src/server.c:1878
#5  beforeSleep (eventLoop=<optimized out>) at /home/ada/Project/valkey/src/server.c:1842
#6  0x00005555555fa1df in aeProcessEvents (flags=27, eventLoop=0x7ffff744d000) at /home/ada/Project/valkey/src/ae.c:426
#7  aeMain (eventLoop=0x7ffff744d000) at /home/ada/Project/valkey/src/ae.c:543
#8  0x00005555555e94fb in main (argc=6, argv=0x7fffffffe298) at /home/ada/Project/valkey/src/server.c:7795

Reading bottom-up shows the call chain clearly — a sharp analysis tool. When node=0x0 above, that is the failure. beforeSleep → connTypeProcessPendingData → rdmaProcessPendingData（rdma.c:1785）→ listDelNode(pending_list, rdma_conn->pending_list_node) → listUnlinkNode, with node == NULL. Multiple frees caused this.

Full circle: this is the same Ctrl+C server crash we fixed in Valkey Over RDMA: server UAF in tests Closed the loop — merging the latest code should kill it: link. Read the fix rationale. Tests no longer hit it — excellent.

Polish the details
#

Looks fixed, but details still need polishing. After the first change we noticed something interesting about this mixed command set:

./src/valkey-benchmark -h 10.0.0.1 -p 6379 -d 256 --threads 16 -c 500 -P 256 -n 50000000 -t set,get --rdma

It should do many SETs first, then many GETs. Before reordering, SET was fine, but the following GET failed — another issue I claimed: https://github.com/valkey-io/valkey/issues/3356 So keep fixing on top of that.

One unclear point: can our new-I/O-model changes apply directly to the old branch, so we do not need two PRs — just keep patching the original? We know the approach; we need a cleaner way.

Probably not — latest unstable and the old tree need different techniques. Try experimental patches first; hard; wait for review comments before pushing further. Land one version for now; revisit if problems remain.

Glance at the new IO model
#

Feature: Redis IO model — here IO threads go from passive to active Architecture comparison Old architecture (polling): Main IO1 IO2 IO3


–[job]–>[queue1]——>
–[job]–>[queue2]——>———–>
–[job]–>[queue3]——>———————>

<-[poll all clients]<-
(inefficient, wastes CPU)

New architecture (event-driven): Main IO1 IO2 IO3

↓ ↓ ↓
–[job]–>[shared SPMC queue]←-compete pull-
(auto load balance)

<—[MPSC response queue]<—–+—-+—-+
(fast notify, no polling)

Still seeing IO hang
#

Do not work around it — if it can hang, the change itself is wrong. htop — inspect memory/CPU; where is the problem? perf — once you have a pid, find hot functions and why it stopped there. Still want to check whether old logic can apply directly. Still cannot find why it hangs.

Record the hang-fix process
#

Currently hangs intermittently. Symptoms:

Usually two CPU cores spin.
After it ends, restarting benchmark hangs immediately again.

Client or server stuck?
#

Use strace — first see what benchmark threads are doing:

pid="$(pgrep -nf valkey-benchmark)"
sudo timeout 15 strace -c -f -p "$pid"

timeout 15 stops after 15s; with per-thread stats you see:

~/Project/valkey fix-rdma-assertion-new-io* ❯ pid="$(pgrep -nf valkey-benchmark)"
sudo timeout 15 strace -c -f -p "$pid"
strace: Process 64770 attached with 17 threads
strace: Process 64791 detached
strace: Process 64793 detached
strace: Process 64792 detached
strace: Process 64790 detached
strace: Process 64789 detached
strace: Process 64788 detached
strace: Process 64787 detached
strace: Process 64786 detached
strace: Process 64785 detached
strace: Process 64784 detached
strace: Process 64783 detached
strace: Process 64782 detached
strace: Process 64781 detached
strace: Process 64780 detached
strace: Process 64779 detached
strace: Process 64778 detached
strace: Process 64770 detached
% time     seconds  usecs/call     calls    errors syscall
------ ----------- ----------- --------- --------- ----------------
 99.89    0.233949         244       956           epoll_wait
  0.11    0.000252           4        59           write
------ ----------- ----------- --------- --------- ----------------
100.00    0.234201         230      1015           total

Benchmark threads all sit in epoll_wait — so the server RDMA path is stuck and stopped replying.

What is the server doing?
#

PID=64539
top - 23:06:29 up  9:08,  1 user,  load average: 2.75, 2.84, 2.81
Threads: 13 total, 2 running, 11 sleep, 0 d-sleep, 0 stopped, 0 zombie
%Cpu(s):  7.7 us,  0.5 sy,  0.0 ni, 91.1 id,  0.5 wa,  0.2 hi,  0.2 si,  0.0 st
MiB Mem :  15673.7 total,   1471.2 free,  12377.8 used,   4286.4 buff/cache
MiB Swap:   4096.0 total,   2244.7 free,   1851.3 used.   3295.9 avail Mem

    PID USER      PR  NI    VIRT    RES    SHR S  %CPU  %MEM     TIME+ COMMAND
  64539 root      20   0 2946372   2.6g   1.5g R  99.3  16.9  40:12.97 valkey-+
  64545 root      20   0 2946372   2.6g   1.5g R  99.3  16.9  40:16.35 io_thd_1
  64540 root      20   0 2946372   2.6g   1.5g S   0.0  16.9   0:00.00 bio_clo+
  64541 root      20   0 2946372   2.6g   1.5g S   0.0  16.9   0:00.00 bio_aof
  64542 root      20   0 2946372   2.6g   1.5g S   0.0  16.9   0:00.00 bio_laz+
  64543 root      20   0 2946372   2.6g   1.5g S   0.0  16.9   0:00.00 bio_rdb+
  64544 root      20   0 2946372   2.6g   1.5g S   0.0  16.9   0:00.00 bio_tls+
  64546 root      20   0 2946372   2.6g   1.5g S   0.0  16.9   0:06.25 io_thd_2
  64547 root      20   0 2946372   2.6g   1.5g S   0.0  16.9   0:05.07 io_thd_3
  64548 root      20   0 2946372   2.6g   1.5g S   0.0  16.9   0:00.00 jemallo+
  64549 root      20   0 2946372   2.6g   1.5g S   0.0  16.9   0:00.00 jemallo+
  64550 root      20   0 2946372   2.6g   1.5g S   0.0  16.9   0:00.00 jemallo+
  64551 root      20   0 2946372   2.6g   1.5g S   0.0  16.9   0:00.00 jemallo+

You should see one main-thread CPU and one IO thread busy-waiting — mainly those two.

Kernel stack snapshot without GDB
#

sudo cat /proc/$pid/stack
for tid in /proc/$pid/task/*; do
  echo "=== $(basename $tid) ==="
  sudo cat $tid/stack 2>/dev/null | head -20
done | less

Walk all tids under the pid and dump stacks. But 64539 and 64545 were not captured above?

Then snapshot with GDB
#

~/Project/valkey fix-rdma-assertion-new-io* ❯ sudo gdb -p "$pid" -batch \
  -ex "set pagination off" \
  -ex "thread apply all bt" \
  -ex "detach" \
  -ex "quit" \
  2>&1 | tee /tmp/valkey-hang-bt.txt

# lightweight process 轻量级进程:这12个是子进程 最后面的64539是主进程
[New LWP 64551]
[New LWP 64550]
[New LWP 64549]
[New LWP 64548]
[New LWP 64547]
[New LWP 64546]
[New LWP 64545]
[New LWP 64544]
[New LWP 64543]
[New LWP 64542]
[New LWP 64541]
[New LWP 64540]
[Thread debugging using libthread_db enabled]
Using host libthread_db library "/usr/lib/libthread_db.so.1".
postponeClientRead (c=0x7f00c6b68e80) at /home/ada/Project/valkey/src/networking.c:6409
6409        if (ProcessingEventsWhileBlocked) return 0;

Thread 13 (Thread 0x7f00ff7326c0 (LWP 64540) "bio_close_file"):
#0  0x00007f0100c9ef32 in ?? () from /usr/lib/libc.so.6
#1  0x00007f0100c9339c in ?? () from /usr/lib/libc.so.6
#2  0x00007f0100c9368c in ?? () from /usr/lib/libc.so.6
#3  0x00007f0100c95e5e in pthread_cond_wait () from /usr/lib/libc.so.6
#4  0x0000558c5f21b00b in mutexQueuePop (theQueue=0x7f0100852380, blocking=<optimized out>) at /home/ada/Project/valkey/src/mutexqueue.c:123                                                                            
#5  0x0000558c5f15ecee in bioProcessBackgroundJobs (arg=0x558c5f4f1280 <bio_workers.lto_priv>) at /home/ada/Project/valkey/src/bio.c:269                                                                                
#6  0x00007f0100c9697a in ?? () from /usr/lib/libc.so.6
#7  0x00007f0100d1a2bc in ?? () from /usr/lib/libc.so.6

Thread 12 (Thread 0x7f00fef316c0 (LWP 64541) "bio_aof"):
#0  0x00007f0100c9ef32 in ?? () from /usr/lib/libc.so.6
#1  0x00007f0100c9339c in ?? () from /usr/lib/libc.so.6
#2  0x00007f0100c9368c in ?? () from /usr/lib/libc.so.6
#3  0x00007f0100c95e5e in pthread_cond_wait () from /usr/lib/libc.so.6
#4  0x0000558c5f21b00b in mutexQueuePop (theQueue=0x7f01008523f0, blocking=<optimized out>) at /home/ada/Project/valkey/src/mutexqueue.c:123                                                                            
#5  0x0000558c5f15ecee in bioProcessBackgroundJobs (arg=0x558c5f4f1298 <bio_workers.lto_priv+24>) at /home/ada/Project/valkey/src/bio.c:269                                                                             
#6  0x00007f0100c9697a in ?? () from /usr/lib/libc.so.6
#7  0x00007f0100d1a2bc in ?? () from /usr/lib/libc.so.6

Thread 11 (Thread 0x7f00fe7306c0 (LWP 64542) "bio_lazy_free"):
#0  0x00007f0100c9ef32 in ?? () from /usr/lib/libc.so.6
#1  0x00007f0100c9339c in ?? () from /usr/lib/libc.so.6
#2  0x00007f0100c9368c in ?? () from /usr/lib/libc.so.6
#3  0x00007f0100c95e5e in pthread_cond_wait () from /usr/lib/libc.so.6
#4  0x0000558c5f21b00b in mutexQueuePop (theQueue=0x7f0100852460, blocking=<optimized out>) at /home/ada/Project/valkey/src/mutexqueue.c:123                                                                            
#5  0x0000558c5f15ecee in bioProcessBackgroundJobs (arg=0x558c5f4f12b0 <bio_workers.lto_priv+48>) at /home/ada/Project/valkey/src/bio.c:269                                                                             
#6  0x00007f0100c9697a in ?? () from /usr/lib/libc.so.6
#7  0x00007f0100d1a2bc in ?? () from /usr/lib/libc.so.6

Thread 10 (Thread 0x7f00fdf2f6c0 (LWP 64543) "bio_rdb_save"):
#0  0x00007f0100c9ef32 in ?? () from /usr/lib/libc.so.6
#1  0x00007f0100c9339c in ?? () from /usr/lib/libc.so.6
#2  0x00007f0100c9368c in ?? () from /usr/lib/libc.so.6
#3  0x00007f0100c95e5e in pthread_cond_wait () from /usr/lib/libc.so.6
#4  0x0000558c5f21b00b in mutexQueuePop (theQueue=0x7f01008524d0, blocking=<optimized out>) at /home/ada/Project/valkey/src/mutexqueue.c:123                                                                            
#5  0x0000558c5f15ecee in bioProcessBackgroundJobs (arg=0x558c5f4f12c8 <bio_workers.lto_priv+72>) at /home/ada/Project/valkey/src/bio.c:269                                                                             
#6  0x00007f0100c9697a in ?? () from /usr/lib/libc.so.6
#7  0x00007f0100d1a2bc in ?? () from /usr/lib/libc.so.6

Thread 9 (Thread 0x7f00fd72e6c0 (LWP 64544) "bio_tls_reload"):
#0  0x00007f0100c9ef32 in ?? () from /usr/lib/libc.so.6
#1  0x00007f0100c9339c in ?? () from /usr/lib/libc.so.6
#2  0x00007f0100c9368c in ?? () from /usr/lib/libc.so.6
#3  0x00007f0100c95e5e in pthread_cond_wait () from /usr/lib/libc.so.6
#4  0x0000558c5f21b00b in mutexQueuePop (theQueue=0x7f0100852540, blocking=<optimized out>) at /home/ada/Project/valkey/src/mutexqueue.c:123                                                                            
#5  0x0000558c5f15ecee in bioProcessBackgroundJobs (arg=0x558c5f4f12e0 <bio_workers.lto_priv+96>) at /home/ada/Project/valkey/src/bio.c:269                                                                             
#6  0x00007f0100c9697a in ?? () from /usr/lib/libc.so.6
#7  0x00007f0100d1a2bc in ?? () from /usr/lib/libc.so.6

Thread 8 (Thread 0x7f00fcf2d6c0 (LWP 64545) "io_thd_1"):
#0  0x0000558c5f21a612 in __rdtsc () at /usr/lib/gcc/x86_64-pc-linux-gnu/15.2.1/include/ia32intrin.h:114
#1  getMonotonicUs_x86 () at /home/ada/Project/valkey/src/monotonic.c:46
#2  0x0000558c5f1dfa66 in IOThreadMain (myid=<optimized out>) at /home/ada/Project/valkey/src/io_threads.c:314                                                                                                          
#3  0x00007f0100c9697a in ?? () from /usr/lib/libc.so.6
#4  0x00007f0100d1a2bc in ?? () from /usr/lib/libc.so.6

Thread 7 (Thread 0x7f00fc72c6c0 (LWP 64546) "io_thd_2"):
#0  0x00007f0100c938d0 in ?? () from /usr/lib/libc.so.6
#1  0x00007f0100c9a0c4 in pthread_mutex_lock () from /usr/lib/libc.so.6
#2  0x0000558c5f1dfbf3 in IOThreadMain (myid=<optimized out>) at /home/ada/Project/valkey/src/io_threads.c:384                                                                                                          
#3  0x00007f0100c9697a in ?? () from /usr/lib/libc.so.6
#4  0x00007f0100d1a2bc in ?? () from /usr/lib/libc.so.6

Thread 6 (Thread 0x7f00fbf2b6c0 (LWP 64547) "io_thd_3"):
#0  0x00007f0100c938d0 in ?? () from /usr/lib/libc.so.6
#1  0x00007f0100c9a0c4 in pthread_mutex_lock () from /usr/lib/libc.so.6
#2  0x0000558c5f1dfbf3 in IOThreadMain (myid=<optimized out>) at /home/ada/Project/valkey/src/io_threads.c:384                                                                                                          
#3  0x00007f0100c9697a in ?? () from /usr/lib/libc.so.6
#4  0x00007f0100d1a2bc in ?? () from /usr/lib/libc.so.6

Thread 5 (Thread 0x7f00fb72a6c0 (LWP 64548) "jemalloc_bg_thd"):
#0  0x00007f0100c9ef32 in ?? () from /usr/lib/libc.so.6
#1  0x00007f0100c9339c in ?? () from /usr/lib/libc.so.6
#2  0x00007f0100c9368c in ?? () from /usr/lib/libc.so.6
#3  0x00007f0100c95e5e in pthread_cond_wait () from /usr/lib/libc.so.6
#4  0x0000558c5f34a8f5 in background_thread_sleep (tsdn=<optimized out>, info=<optimized out>, interval=<optimized out>) at src/background_thread.c:137                                                                 
#5  background_work_sleep_once (tsdn=<optimized out>, info=<optimized out>, ind=0) at src/background_thread.c:229                                                                                                       
#6  background_thread0_work (tsd=<optimized out>) at src/background_thread.c:374
#7  background_work (tsd=<optimized out>, ind=<optimized out>) at src/background_thread.c:412
#8  background_thread_entry (ind_arg=<optimized out>) at src/background_thread.c:444
#9  0x00007f0100c9697a in ?? () from /usr/lib/libc.so.6
#10 0x00007f0100d1a2bc in ?? () from /usr/lib/libc.so.6

Thread 4 (Thread 0x7f00fa7ff6c0 (LWP 64549) "jemalloc_bg_thd"):
#0  0x00007f0100c9ef32 in ?? () from /usr/lib/libc.so.6
#1  0x00007f0100c9339c in ?? () from /usr/lib/libc.so.6
#2  0x00007f0100c9368c in ?? () from /usr/lib/libc.so.6
#3  0x00007f0100c95e5e in pthread_cond_wait () from /usr/lib/libc.so.6
#4  0x0000558c5f34a239 in background_thread_sleep (tsdn=<optimized out>, info=0x7f0100a16950, interval=<optimized out>) at src/background_thread.c:137                                                                  
#5  background_work_sleep_once (tsdn=<optimized out>, info=<optimized out>, ind=<optimized out>) at src/background_thread.c:229                                                                                         
#6  background_work (tsd=<optimized out>, ind=<optimized out>) at src/background_thread.c:419
#7  background_thread_entry (ind_arg=<optimized out>) at src/background_thread.c:444
#8  0x00007f0100c9697a in ?? () from /usr/lib/libc.so.6
#9  0x00007f0100d1a2bc in ?? () from /usr/lib/libc.so.6

Thread 3 (Thread 0x7f00f9ffe6c0 (LWP 64550) "jemalloc_bg_thd"):
#0  0x00007f0100c9ef32 in ?? () from /usr/lib/libc.so.6
#1  0x00007f0100c9339c in ?? () from /usr/lib/libc.so.6
#2  0x00007f0100c9368c in ?? () from /usr/lib/libc.so.6
#3  0x00007f0100c95e5e in pthread_cond_wait () from /usr/lib/libc.so.6
#4  0x0000558c5f34a239 in background_thread_sleep (tsdn=<optimized out>, info=0x7f0100a16a20, interval=<optimized out>) at src/background_thread.c:137                                                                  
#5  background_work_sleep_once (tsdn=<optimized out>, info=<optimized out>, ind=<optimized out>) at src/background_thread.c:229                                                                                         
#6  background_work (tsd=<optimized out>, ind=<optimized out>) at src/background_thread.c:419
#7  background_thread_entry (ind_arg=<optimized out>) at src/background_thread.c:444
#8  0x00007f0100c9697a in ?? () from /usr/lib/libc.so.6
#9  0x00007f0100d1a2bc in ?? () from /usr/lib/libc.so.6

Thread 2 (Thread 0x7f00f93ff6c0 (LWP 64551) "jemalloc_bg_thd"):
#0  0x00007f0100c9ef32 in ?? () from /usr/lib/libc.so.6
#1  0x00007f0100c9339c in ?? () from /usr/lib/libc.so.6
#2  0x00007f0100c9368c in ?? () from /usr/lib/libc.so.6
#3  0x00007f0100c95e5e in pthread_cond_wait () from /usr/lib/libc.so.6
#4  0x0000558c5f34a239 in background_thread_sleep (tsdn=<optimized out>, info=0x7f0100a16af0, interval=<optimized out>) at src/background_thread.c:137                                                                  
#5  background_work_sleep_once (tsdn=<optimized out>, info=<optimized out>, ind=<optimized out>) at src/background_thread.c:229                                                                                         
#6  background_work (tsd=<optimized out>, ind=<optimized out>) at src/background_thread.c:419
#7  background_thread_entry (ind_arg=<optimized out>) at src/background_thread.c:444
#8  0x00007f0100c9697a in ?? () from /usr/lib/libc.so.6
#9  0x00007f0100d1a2bc in ?? () from /usr/lib/libc.so.6

Thread 1 (Thread 0x7f0100e32740 (LWP 64539) "valkey-server"):
#0  postponeClientRead (c=0x7f00c6b68e80) at /home/ada/Project/valkey/src/networking.c:6409
#1  readQueryFromClient (conn=0x7f00f3035f40) at /home/ada/Project/valkey/src/networking.c:4256
#2  0x0000558c5f260a9a in callHandler (conn=0x7f00f3035f40, handler=<optimized out>) at /home/ada/Project/valkey/src/connhelpers.h:79                                                                                   
#3  connRdmaEventHandler (el=<optimized out>, fd=<optimized out>, clientData=0x7f00f3035f40, mask=<optimized out>) at /home/ada/Project/valkey/src/rdma.c:735                                                           
#4  0x0000558c5f22d3b5 in connUpdateState (conn=<optimized out>) at /home/ada/Project/valkey/src/connection.h:524                                                                                                       
#5  processClientIOWriteDone (c=0x7f00c6b68e80) at /home/ada/Project/valkey/src/networking.c:3215
#6  processClientIOWriteDone (c=0x7f00c6b68e80) at /home/ada/Project/valkey/src/networking.c:3204
#7  0x0000558c5f1dfec7 in handleWriteJobs (write_jobs=0x7fff24438e80, write_count=<optimized out>) at /home/ada/Project/valkey/src/io_threads.c:884                                                                     
#8  processIOThreadsResponses () at /home/ada/Project/valkey/src/io_threads.c:934
#9  0x0000558c5f29d228 in processIOThreadsResponses () at /home/ada/Project/valkey/src/io_threads.c:73
#10 beforeSleep (eventLoop=<optimized out>) at /home/ada/Project/valkey/src/server.c:1975
#11 beforeSleep (eventLoop=<optimized out>) at /home/ada/Project/valkey/src/server.c:1842
#12 0x0000558c5f14f1df in aeProcessEvents (flags=27, eventLoop=0x7f010084d000) at /home/ada/Project/valkey/src/ae.c:426                                                                                                 
#13 aeMain (eventLoop=0x7f010084d000) at /home/ada/Project/valkey/src/ae.c:543
#14 0x0000558c5f13e4fb in main (argc=6, argv=0x7fff24439388) at /home/ada/Project/valkey/src/server.c:7795
[Inferior 1 (process 64539) detached]

Read bottom-up (from outer frames into the core). At #0 the thread stopped here:

Try to fix the root cause: It is a state-machine misalignment, already written into this fix busy-spin commit message — reread. Classic state-machine confusion.

Semantic-level refactor
#

What refactor approaches do we have?

Root cause: before offload, set postpone; after finish, clear it; then updateEvent.

Unused logic: postpone_update_state has an unused parameter. Triggered from two paths — maybe usable. Normal events only enter on POLLIN (only POLLIN registered?). Mask only allows POLLIN in. When we enter via the handler above, other parameters can control whether read/write handlers run; with mask we know the trigger path: either system (RDMA CQ) POLLIN via aeEvent — mask is only POLLIN — or when postpone clears and updateState runs, use a dedicated mask on that path and see how the error fires. Or reserve a mask bit for internal use — decide whether to wake specific handlers. Might fix it. Avoid our extra int flag; postpone came from TLS; RDMA does not want to dirty networking, but worth trying.

Walk the root cause again
#

Walk the code once more. Mindset while fixing: main and IO threads communicate via lock-free queues. Root cause is clear; what remains is whether the fix is elegant.

1. Main-thread state
#

Start from the read callback (reentrancy is in the read path): readQueryFromClient:

void readQueryFromClient(connection *conn) {
	client *c = connGetPrivateData(conn);
	/* Check if we can send the client to be handled by the IO-thread */
	// 尝试能否推迟ClientRead
	if (postponeClientRead(c)) return;
	if (c->io_write_state != CLIENT_IDLE || c->io_read_state != CLIENT_IDLE) return;
	......
	}

Calls postponeClientRead: (postpone = can we hand this to an IO thread for now)

/* Return 1 if the client read is handled using threaded I/O.
* 0 otherwise. */
int postponeClientRead(client * c) {
    if (ProcessingEventsWhileBlocked) return 0;
    // 尝试看这个client read是否能够派发给IO线程来进行处理。
    return (trySendReadToIOThreads(c) == C_OK);
}

Then trySendReadToIOThreads, matching the comments:

int trySendReadToIOThreads(client * c) {
	// 这里有自己的计算方法
    if (server.active_io_threads_num <= 1) return C_ERR;
    /* Fake/teardown clients may have no connection; never offload those. */
    if (!c - > conn) return C_ERR;
    /* If IO thread is still reading, return C_OK so the main thread does not race it. */
    if (c - > io_read_state == CLIENT_PENDING_IO) return C_OK;
    /* A completed read must be finished by processClientIOReadsDone on the main thread
     * before we try to offload another read; do not treat it like PENDING_IO. */
       
    // 这里返回C_ERR来解决状态机紊乱的问题。 目前的状态是C
    
    if (c - > io_read_state == CLIENT_COMPLETED_IO) return C_ERR;
    if (c - > io_write_state == CLIENT_PENDING_IO) return C_OK;
    /* For simplicity, don't offload replica clients reads as read traffic from replica is negligible */
    if (getClientType(c) == CLIENT_TYPE_REPLICA) return C_ERR;
    /* With Lua debug client we may call connWrite directly in the main thread */
    if (c - > flag.lua_debug) return C_ERR;
    /* For simplicity let the main-thread handle the blocked clients */
    if (c - > flag.blocked || c - > flag.unblocked) return C_ERR;
    if (c - > flag.close_asap) return C_ERR;

    c - > read_flags = canParseCommand(c) ? 0 : READ_FLAGS_DONT_PARSE;
    c - > read_flags |= authRequired(c) ? READ_FLAGS_AUTH_REQUIRED : 0;
    c - > read_flags |= isReplicatedClient(c) ? READ_FLAGS_REPLICATED : 0;
    c - > io_read_state = CLIENT_PENDING_IO;
    connSetPostponeUpdateState(c - > conn, 1);
	
	// 单生产者多消费者队列
    if (unlikely(spmcEnqueue( & io_shared_inbox, tagJob(c, JOB_REQ_READ_CLIENT)) == false)) {
        c - > read_flags = 0;
        c - > io_read_state = CLIENT_IDLE;
        connSetPostponeUpdateState(c - > conn, 0);
        return C_ERR;
    }

	// 此时真的offload当前这个readjob
    io_jobs_submitted++;
    server.stat_io_reads_pending++;
    c - > flag.pending_read = 1;
    return C_OK;
}

After a successful return, this client read should already be offloaded to an IO thread. Note two points (keep analyzing main on this line; or jump to client code below):

c -> io_read_state = CLIENT_PENDING_IO;
connSetPostponeUpdateState(c -> conn, 1);

We set the first read state machine to CLIENT_PENDING_IO, and call connSetPostponeUpdateState; if that function pointer is registered, we invoke the callback:

static inline void connSetPostponeUpdateState(connection *conn, int on) {
	if (conn->type->postpone_update_state) {
		conn->type->postpone_update_state(conn, on);
	}
}

The callback pointer is defined in connection.c:

typedef struct connectionType {
	/* Postpone update state - with IO threads & TLS we don't want the IO threads to update the event loop events - let
* the main-thread do it */
  // 这个回调指针是TLS引入的，就是说在TLS连接层和开启IO线程的情况下，我们不希望IO线程来更新 事件循环事件（这是什么？）而是使用主线程来进行更新。
	void (*postpone_update_state)(struct connection *conn, int);
}

What does postpone_update_state do? Only RDMA and TLS use this callback:

static ConnectionType CT_RDMA = {
......
.postpone_update_state = postPoneUpdateRdmaState, 
......
}

That is the concrete function — look at postPoneUpdateRdmaState:

static void postPoneUpdateRdmaState(struct connection *conn, int postpone) {
rdma_connection *rdma_conn = (rdma_connection *)conn;
if (postpone) {
	rdma_conn->flags |= RDMA_CONN_FLAG_POSTPONE_UPDATE_STATE;
} else {
	rdma_conn->flags &= ~RDMA_CONN_FLAG_POSTPONE_UPDATE_STATE;
}
}

If postpone is needed (nonzero arg → set RDMA_CONN_FLAG_POSTPONE_UPDATE_STATE, else clear it). Where is the flag used? Within this connection we can always operate on it. (Actually used after IO’s sendToMainThread, so continue from _main receiving IO replies.)

#define RDMA_CONN_FLAG_POSTPONE_UPDATE_STATE (1 << 0)

The define above is 1.

============ divider ============

`_main` after IO returns results
#

beforeSleep runs each round before 「preparing to block waiting for I/O」: Readable/writable callbacks from the previous poll have finished; beforeSleep finishes this phase’s bookkeeping, then enters poll (or custompoll / IO-thread poll).

Watch processIOThreadsResponses() inside beforeSleep: What is it doing? Pull completed read/write/accept results from IO threads back to main: advance networking/client state.

int processIOThreadsResponses(void) {

    /* We don't check for threads number since some threads may return jobs then deactivate/shut-down */
    /* Quick check if any pending operations exist */
    if (getPendingIOResponsesCount() == 0) return 0;
    int total_processed = 0;
    void * jobs[JOB_BATCH_SIZE];
    client * read_jobs[JOB_BATCH_SIZE];
    client * write_jobs[JOB_BATCH_SIZE];

    /* Loop until we consume all pending jobs */
    // 循环遍历处理之前IO线程返回的每一个JOB
    while (1) {
        int received_responses = 0;
        int dequeued_count = 0;
        int read_count = 0;
        int write_count = 0;
        /* Try to dequeue JOB_BATCH_SIZE */
        while (received_responses < JOB_BATCH_SIZE) {

            // 从多生产者单消费者队列中取jobs进行处理
            // 批量进行出队的操作
            dequeued_count = mpscDequeueBatch( & io_shared_outbox, jobs, JOB_BATCH_SIZE - received_responses);
            /* Stop if we can't get more jobs from the queue. */

            if (dequeued_count == 0) break;
            received_responses += dequeued_count;
            total_processed += dequeued_count;
            
            for (int i = 0; i < dequeued_count; i++) {
                void * data;
                int job_type;
                // 这里把client给解出来
                untagJob(jobs[i], & data, & job_type);
                client * c = (client * ) data;
                
                if (job_type == JOB_RES_READ_CLIENT) {
                
                    // 此时的状态必须是IO处理结束的
                    serverAssert(c - > io_read_state == CLIENT_COMPLETED_IO);
                    read_jobs[read_count++] = c;
                } else if (job_type == JOB_RES_WRITE_CLIENT) {
                
                    serverAssert(c - > io_write_state == CLIENT_COMPLETED_IO);
                    write_jobs[write_count++] = c;
                } else {
                    serverPanic("Unknown job type %d", job_type);
                }
            }
        }
        // 对于要读和要写的操作调用不同的函数。
        if (read_count) handleReadJobs(read_jobs, read_count);
        if (write_count) handleWriteJobs(write_jobs, write_count);
        /* If the queue was empty at the last try - don't try again */
        if (dequeued_count == 0) return total_processed;
    }
}

It walks many jobs; IO enqueued read jobs (our whole path is read analysis); for the collected read-jobs array call handleReadJobs; look at that function (unmodified version — we are still sorting root cause):

/* Function to handle read jobs */
static void handleReadJobs(client ** read_jobs, int read_count) {
	// 我们这次处理了这么多读任务
    server.stat_io_reads_pending -= read_count;
    serverAssert(server.stat_io_reads_pending >= 0);

    /* process each client */
    // 遍历每个client客户端
    for (int i = 0; i < read_count; i++) {
        client * c = read_jobs[i];
        processClientIOReadsDone(c);
    }

    /* Process commands in batch if we processed any reads */
    if (read_count) {
        server.stat_io_reads_processed += read_count;
        processClientsCommandsBatch();
    }
}

Old logic: walk each collected client, then processClientIOReadsDone — what does that do?

This should already be the wrap-up.

void processClientIOReadsDone(client * c) {
    serverAssert(c - > io_read_state == CLIENT_COMPLETED_IO);

    if (ProcessingEventsWhileBlocked) {
        /* When ProcessingEventsWhileBlocked we may call processIOThreadsReadDone recursively.
         * In this case, there may be some clients left in the batch waiting to be processed. */
        processClientsCommandsBatch();
    }

    c - > flag.pending_read = 0;
    c - > io_read_state = CLIENT_IDLE;
    /* Don't post-process-reads from clients that are going to be closed anyway. */
    if (c - > flag.close_asap) return;
    /* If a client is protected, don't do anything,
     * that may trigger read/write error or recreate handler. */
    if (c - > flag.protected) return;

    /* Save the current conn state, as connUpdateState may modify it */
    int in_accept_state = (connGetState(c - > conn) == CONN_STATE_ACCEPTING);
    connSetPostponeUpdateState(c - > conn, 0);
    connUpdateState(c - > conn);

    /* In accept state, no client's data was read - stop here. */
    if (in_accept_state) return;
    /* On read error - stop here. */
    if (handleReadResult(c) == C_ERR) {
        return;
    }

    if (!(c - > read_flags & READ_FLAGS_DONT_PARSE)) {
        parseResult res = handleParseResults(c);
        /* On parse error - stop here. */
        if (res == PARSE_ERR) {
            return;
        } else if (res == PARSE_NEEDMORE) {
            beforeNextClient(c);
            return;
        }
    }
    if (c - > argc > 0) {
        c - > flag.pending_command = 1;
    }
    /* try to add the command to the batch */
    int ret = addCommandToBatchAndProcessIfFull(c);
    /* If the command was not added to the commands batch, process it immediately */
    if (ret == C_ERR) {
        if (processPendingCommandAndInputBuffer(c) == C_OK) beforeNextClient(c);
    }
}

Note: client state becomes IDLE — idle again, reset.

  c - > flag.pending_read = 0;
  c - > io_read_state = CLIENT_IDLE;
  /* Don't post-process-reads from clients that are going to be closed anyway. */
  if (c - > flag.close_asap) return;

State update — this is where it breaks:

    /* Save the current conn state, as connUpdateState may modify it */
    int in_accept_state = (connGetState(c - > conn) == CONN_STATE_ACCEPTING);
    connSetPostponeUpdateState(c - > conn, 0);
    connUpdateState(c - > conn);

After finishing, we restore the postponeUpdateState function pointer. In processClientIOReadsDone we update state via RDMA’s connUpdateState:

static inline void connUpdateState(connection *conn) {
	if (conn->type->update_state) {
		conn->type->update_state(conn);
	}
}

That calls the registered update_state callback:

static ConnectionType CT_RDMA = {
......
.postpone_update_state = postPoneUpdateRdmaState,
.update_state = updateRdmaState,
......
};

Look at updateRdmaState:

static void updateRdmaState(struct connection * conn) {
    rdma_connection * rdma_conn = (rdma_connection * ) conn;
    connRdmaSetRwHandler(conn);
    connRdmaEventHandler(NULL, -1, rdma_conn, 0);
}

connRdmaSetRwHandler only registers the readable handler — IB can only register POLLIN. Then connRdmaEventHandler runs:

static void connRdmaEventHandler(struct aeEventLoop * el, int fd, void * clientData, int mask) {
    rdma_connection * rdma_conn = (rdma_connection * ) clientData;
    connection * conn = & rdma_conn - > c;
    struct rdma_cm_id * cm_id = rdma_conn - > cm_id;
    RdmaContext * ctx = cm_id - > context;

    int ret = 0;

    UNUSED(el);
    UNUSED(fd);
    UNUSED(mask);

    ret = connRdmaHandleCq(rdma_conn);
    if (ret == C_ERR) {
        conn - > state = CONN_STATE_ERROR;
        return;
    }

    /* uplayer should read all */
    while (!(rdma_conn - > flags & RDMA_CONN_FLAG_POSTPONE_UPDATE_STATE) && ctx - > rx.pos < ctx - > rx.offset) {
        if (conn - > read_handler && (callHandler(conn, conn - > read_handler) == C_ERR)) {
            return;
        }
    }

    /* recv buf is full, register a new RX buffer */
    if (ctx - > rx.pos == ctx - > rx.length) {
        connRdmaRegisterRx(ctx, cm_id);
    }

    /* RDMA comp channel has no POLLOUT event, try to send remaining buffer */
    if (!(rdma_conn - > flags & RDMA_CONN_FLAG_POSTPONE_UPDATE_STATE) && ctx - > tx.offset < ctx - > tx.length && conn - > write_handler) {
        callHandler(conn, conn - > write_handler);
    }
}

During this state update it also callHandlers into the registered read_handler. Who is read_handler? readQueryFromClient — reentrancy: walk the same path again down to:

void parseInputBuffer(client *c) {
	/* The command queue must be emptied before parsing. */
	// 指令队列在解析之前必须为空
	serverAssert(c->cmd_queue.len == 0);
	......
}

Reentered while prior work may not be finished → assertion (my inference). processClientIOWriteDone also calls update and enters the read path, causing another:

/* If io_last_written_data_len is nonzero it must relate to c->buf */
serverAssert(c->io_last_written.data_len == 0 || c->io_last_written.buf == c->buf);

assertion failure — this PR tries to fix both.

IO-thread state
#

The two state machines interleave — analyze alternately. Now I am the IO thread.

After main offloaded the client read to us, walk down from the IO main:

static void * IOThreadMain(void * myid) {
        /* The ID is the thread ID number (from 1 to server.io_threads_num-1). ID 0 is the main thread. */
        long id = (long) myid;
        char thdname[32];
        ......
        while (1) {
            ......
            /* PRIORITY 2: Shared Global Queue (SPMC)
             * Only checked after SPSC is drained. */
			 // 这里就是经典的从环形缓冲区尝试进行出队的操作，单生产者多消费者。
            void * tagged_job = spmcDequeue( & io_shared_inbox);
            // 看需要消费的job类型
            if (tagged_job) {
                void * data;
                int type;
                untagJob(tagged_job, & data, & type);
                switch (type) {
                // 
                    case JOB_REQ_READ_CLIENT:
                        ioThreadReadQueryFromClient((client * ) data);
                        break;
                    case JOB_REQ_WRITE_CLIENT:
                        ioThreadWriteToClient((client * ) data);
                        break;
                    case JOB_REQ_FREE_OBJ:
                        zfree(data);
                        break;
                    case JOB_REQ_ACCEPT:
                        ioThreadAccept((client * ) data);
                        break;
                    case JOB_REQ_POLL:
                        ioThreadPoll((aeEventLoop * ) data);
                        break;
                    default:
                        serverPanic("Invalid SPMC job type: %d", type);
                }
                processed++;
                ......
            } {

When _main enqueued, the job tag was JOB_REQ_READ_CLIENT (code path for enqueue failure):

    if (unlikely(spmcEnqueue( & io_shared_inbox, tagJob(c, JOB_REQ_READ_CLIENT)) == false)) {
        c - > read_flags = 0;
        c - > io_read_state = CLIENT_IDLE;
        connSetPostponeUpdateState(c - > conn, 0);
        return C_ERR;
    }

OK, continue: in ioMain’s switch we call ioThreadReadQueryFromClient:

void ioThreadReadQueryFromClient(client * c) {
	// 断言，刚进这个函数的时候，我们的状态肯定是 `CLIENT_PENDING_IO` 就是等待处理的状态。
    serverAssert(c-> io_read_state == CLIENT_PENDING_IO);

    /* Read */
    readToQueryBuf(c);
    if (c - > flag.close_asap) {
        goto done;
    }

    /* Check for read errors. */
    if (c - > nread <= 0) {
        goto done;
    }

    /* Skip command parsing if the READ_FLAGS_DONT_PARSE flag is set. */
    if (c - > read_flags & READ_FLAGS_DONT_PARSE) {
        goto done;
    }

    /* Handle QB limit */
    if (c - > read_flags & READ_FLAGS_QB_LIMIT_REACHED) {
        goto done;
    }
	
	// ======== 这个解析就是重入点，之前的指令主线程还没处理结束 ========
    parseInputBuffer(c);
    
    trimCommandQueue(c);
    prepareCommandQueue(c);

    /* Parsing was not completed - let the main-thread handle it. */
    if (!(c - > read_flags & READ_FLAGS_PARSING_COMPLETED)) {
        goto done;
    }

    /* Empty command - Multibulk processing could see a <= 0 length. */
    if (c - > argc == 0) {
        goto done;
    }

    done:
        /* Only trim query buffer for non-primary clients
         * Primary client's buffer is handled by main thread using repl_applied position */
        if (!(c - > read_flags & READ_FLAGS_REPLICATED)) {
            trimClientQueryBuffer(c);
        }

    c - > io_read_state = CLIENT_COMPLETED_IO;
    c - > cur_tid = getCurTid();
    sendToMainThread(c, JOB_RES_READ_CLIENT);
}

Here we read into the buffer and parse (study later; not the focus), and after finishing the read client we set the state machine to CLIENT_COMPLETED_IO — IO is done; return the result to main:

c - > io_read_state = CLIENT_COMPLETED_IO;
c - > cur_tid = getCurTid();
sendToMainThread(c, JOB_RES_READ_CLIENT);

What does sendToMainThread do?

void sendToMainThread(void * data, int type) {
    if (unlikely(pending_io_responses)) {
        flushPendingIOResponses(0);
    }

	// 拿到我们IO当前完成的任务
    void * job = tagJob(data, type);
    // 把我们当前完成的job放进一个队列中，这是所有IO都放到一个队列内部，然后主线程进行对应的处理。
    // 所以就是多生产者(IO线程)单消费者(_main线程)的链表
    if (unlikely(pending_io_responses || !mpscEnqueue( & io_shared_outbox, job, & io_thread_ticket))) {
        /* Failed to push new job: initialize list if needed and save job */
        // 实在没有放进去的话我们就开启一个全新的队列（这个处理机制是什么???）
        if (pending_io_responses == NULL) {
            pending_io_responses = listCreate();
        }
        listAddNodeTail(pending_io_responses, job);
    }
}

OK — IO’s job is done: after the read client, enqueue the result, then back to _main — back to main

A more elegant fix?
#

Root cause analyzed; want a cleaner fix — upper layers look dirty.
You should not write `if (conn type == RDMA) { ... }` — ugly. Is there a cleaner approach?

Idea: instead of type checks upstairs, use postpone_update_state more objectively; but many edits, easy to get wrong — needs more study; logic is too complex. Prefer a minimal change series, not one big rewrite. Such fixes are never one-shot OK — inherently complex.

When showing AI or needing an unpaginated git diff:

$ git --no-pager -diff # 这个是你可以展示目前工作区的diff

If already committed, how to view:

$ git --no-pager show <commit_id> # 不分页展示某个已经提交的commit

Both are very useful. Looks about done — from February when IO jobs were not yet MPMC (I fixed the old IO bugs), then the IO model changed and I fixed the new one; somehow it is late May — time flies; also fixed two smaller issues. Hope merge goes smoothly!

Timeline after our change is roughly:

sequenceDiagram
    participant Epoll as RDMA/epoll
    participant Main as 主线程
    participant IO as IO 线程
    participant RDMA as connRdmaEventHandler

    Epoll->>Main: readQueryFromClient
    Main->>IO: trySendReadToIOThreads → inbox
    Note over Main: PENDING_IO + postpone READ(这里推迟了读路径)

    IO->>IO: ioThreadReadQueryFromClient<br/>parseInputBuffer → cmd_queue
    IO->>Main: COMPLETED_IO → outbox

    Main->>Main: handleReadJobs ① processClientIOReadsDone
    Note over Main: IDLE, mask=READ(+WRITE?), connUpdateState
    Note over RDMA: READ 被挡，不进 read_handler

    Main->>Main: ② processClientsCommandsBatch
    Main->>Main: ③ drain + clientConnPostponeMask + connUpdateState
    Note over RDMA: queue 空 → 可排空 RX

Essence of the PR: at specific points in the client/connection lifecycle, set READ/WRITE bits in postpone_mask; check those bits in RDMA’s connRdmaEventHandler, deciding 「whether this moment may sync-call read/write handlers」.

CI failure
#

On this branch we hit a stable CI failure: the single-io_threads case in test-ubuntu-latest-cmake-tls. My bug or the test’s? Try to fix — first reproduce the test:

# 使用 CMake 编译带 TLS 支持的 Valkey (与 CI 环境保持一致) 
cmake -B build -DUSE_TLS=ON 
cmake --build build -j$(nproc)

Looks like postpone-mask issues remain — for TLS, what behaves wrongly?

Basic test flow
#

pkill -9 -f valkey-server; pkill -9 -f runtest # 首先杀掉原来的进程
make BUILD_TLS=yes -j$(nproc) # 带TLS进行编译
./utils/gen-test-certs.sh     # 生成相应的证书
./runtest --single tests/unit/io-threads.tcl --tls # 跑这个报错的单测

To match test-ubuntu-latest-cmake-tls exactly, our flow is:

rm -rf build-release
mkdir -p build-release
cd build-release
cmake -DCMAKE_BUILD_TYPE=Release .. -DBUILD_TLS=yes -DBUILD_UNIT_GTESTS=yes
make -j$(nproc)
cd ..

On TLS accept you must not seal the read path (fix the if) — otherwise the TLS state machine cannot advance. After that, CI all green.

Then more review:

processClientIOWriteDone(c)
  → connUpdateState(c->conn)
    → updateRdmaState()
      → connRdmaEventHandler()
        → read handler
          → readQueryFromClient()
            → ... parse/execute ...
              → beforeNextClient(c)
                → if (c->flag.close_asap) freeClient(c);  // c is freed
  → come back to processClientIOWriteDone，try to access c  // use-after-free

On this write path,

also look up by ID first, then check existence — prevent the same UAF.

Ping viktor again:

@zuiderkwast Gentle ping on this one when you have bandwidth — it's in the Valkey 9.1 board under **Needs Review**.

Quick recap since the last round of feedback:
- RDMA + IO threads re-entrancy fix via directional `CONN_POSTPONE_READ` / `CONN_POSTPONE_WRITE` (approved by @pizhenwei).
- `handleReadJobs` drains `cmd_queue` before resuming transport.
- Latest follow-ups: removed blanket RDMA postpone skips (950c32b), inlined the read-done postpone mask (5586e73), and addressed the write-path UAF with post-`connUpdateState()` re-lookup (f842d16).

CI is green on the latest commit. No rush at all — just bubbling it up in case it got buried. Thanks!

As of writing, still not merged — but the analysis is clear.

Reintroducing an interesting problem (background) #

Testing under the new I/O model #

Keep trying to fix it #

Detailed discussion #

While fixing, GET also hits an access-nil error #

Polish the details #

Glance at the new IO model #

Still seeing IO hang #

Record the hang-fix process #

Client or server stuck? #

What is the server doing? #

Kernel stack snapshot without GDB #

Then snapshot with GDB #

Semantic-level refactor #

Walk the root cause again #

1. Main-thread state #

_main after IO returns results #

IO-thread state #

A more elegant fix? #

CI failure #

Basic test flow #