我就直接认为这是一个新的feature了，就是利用libvalkey的client客户端进行模拟的压测，同时是需要支持多线程的。
本次修改的测试逻辑 #
这算是一个新的feature,我们需要提一个新的PR来解决问题。
在tests/rdma的目录下方，增加libvalkey-test.c，直接使用libvalkey的client相关的API来进行测试即可。
我们先研究一下这里的测试逻辑.
tests/rdma 测试的逻辑主要写在这里，但是明显不是常规的测试:
目录结构:
| 文件 | 作用 |
|---|---|
run.py |
主编排：编译、找 RDMA 网卡、起 valkey-server、跑 rdma-test、清理 |
rdma_env.py |
root 下 setup/cleanup 软 RDMA（RXE 模块、rdma link add） |
rdma-test.c |
C 测试客户端：自己实现 RDMA CM/verbs 协议，发 RESP 命令 |
Makefile |
编译 rdma-test（链接 librdmacm、libibverbs） |
这里应该是类似于冒烟测试,pizhenwei直接又手写了一个RDMA的小客户端进行测试。
但是明显没有打开IO多线程。
还是一样先build一次: AI先实现完了，我们看看是怎么实现的？
make -j$(nproc) BUILD_RDMA=yes USE_FAST_FLOAT=yes
make -C tests/rdma
笼统的来讲，就是利用
libvalkey下的client 相关的API来类似模拟我们的这条压力测试命令,然后再修改一下构建的文件，但是代码量有点大，需要理解。
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
我们在原来的分支上直接进行了测试和修改，做了很多commit,但是我想单独把这个测试放在一个PR里，就是撤销这个分支上迄今为止的部分测试，然后重放在unstable分支上面证明存在问题。
你可以自己进行参数的调整并且复现:
~/Project/valkey rdma-libvalkey-test* ❯ sudo ./runtest-rdma --install-rxe \
--io-threads 8 \
-c 500 -P 512 -n 1000000 --timeout 600
这种条件下一定会导致断言错误。
所以需要稍微上调一下，但是又不能太高，防止时间过长或者机器OOM 总之是上升了一些参数
libvalkey-test.c这个测试在干什么？
#
你可以这样理解：用官方 libvalkey 客户端，在 RDMA 上复现 valkey-benchmark 式的 pipeline 压测。
这个测试在验证什么？ #
对比同目录的 rdma-test.c：
| 功能/文件名 | rdma-test |
libvalkey-test |
|---|---|---|
| 客户端栈 | 手写 RDMA CM / CQ / QP | libvalkey（deps/libvalkey） |
| 协议 | 自己拼 RESP、自己读 RDMA 缓冲 | libvalkey 格式化 RESP + RDMA 传输 |
| 并发 | 4 线程，每线程 1 连接 | 16 worker × 每 worker 多个连接 |
| 模式 | 逐条 SET/GET | pipeline：先 append 一批，再 flush，再 drain |
| 目标 | 功能冒烟（PING/SET/GET/BGSAVE） | RDMA + IO thread + 高 pipeline 下的稳定性 |
你要修的 bug（cmd_queue.len == 0 assert）需要同时满足：
- 很多连接同时 inflight 命令（pipeline）
- Server 开 IO thread 读路径（
run.py第二阶段才开）
这个程序就是为此设计的 客户端侧负载生成器。
整体架构 #
flowchart TB
subgraph main["main()"]
init[valkeyInitiateRdma]
spawn[创建 16 个 pthread]
end
subgraph worker["worker_main() × 16"]
conn[每 worker 建 N 个 valkeyContext RDMA 连接]
set[run_phase SET]
get[run_phase GET]
end
subgraph per_conn["每个 client_state / valkeyContext"]
obuf[obuf: 待发 RESP 命令队列]
reader[reader: 已收 RESP 解析队列]
end
main --> worker
worker --> per_conn
obuf -->|valkeyBufferWrite → RDMA| server[valkey-server RDMA]
server -->|RDMA 响应| reader
这里的基本逻辑就是开很多线程，然后每个线程还同时管理很多连接。
注意此处线程和连接的划分:
for (int i = 0; i < cfg.threads; i++) {
workers[i].cfg = & cfg; // 设置当前worker的配置
workers[i].thread_id = i; // 这是第几个线程
workers[i].first_client = (cfg.clients * i) / cfg.threads; // 本线程对应的第一条连接
workers[i].last_client = (cfg.clients * (i + 1)) / cfg.threads; // 本线程对应的最后一条连接
if (pthread_create( & threads[i], NULL, worker_main, & workers[i])) { // 根据我们刚才的设置来创建这个线程，然后进行处理
fprintf(stderr, "failed to create thread %d\n", i);
ret = 1;
cfg.threads = i;
break;
}
}
分配总的请求量：
static long long requests_for_client(const test_config *cfg, int client_id) {
long long base = cfg->requests / cfg->clients;
long long extra = client_id < (cfg->requests % cfg->clients) ? 1 : 0;
return base + extra;
}
不能整除时，前 requests % clients 个 client 各多 1 条，保证总和精确等于 cfg->requests。
设计的核心数据结构 #
typedef struct test_config { // 当前test的配置 全局参数，所有worker read only
const char *host;
int port;
int threads;
int clients;
int pipeline;
long long requests;
size_t datasize;
char *value;
} test_config;
typedef struct worker_config { // 当前工作线程worker的配置
const test_config *cfg;
int thread_id;
int first_client;
int last_client;
} worker_config; // 这里会记录连接线程的区间 以及线程本身的配置
typedef struct client_state { // 一个连接的状态
int client_id; // 客户端ID
long long requests; // 一个连接处理的请求数量
long long processed;
int pending;
valkeyContext *context;
} client_state;
| 字段 | 含义 |
|---|---|
test_config |
全局参数，所有 worker 只读 |
worker_config |
本 pthread 负责的 client 区间 [first_client, last_client) |
client_state.processed |
该连接已完成 SET（或 GET）的条数 |
client_state.pending |
当前 pipeline 批次里已 append、尚未 drain 回复的命令数 |
client_state.context |
libvalkey 连接；内部有 obuf（写缓冲）和 reader（读解析器） |
同时，针对每个连接使用不同的key来测试，防止竞争:
static void make_key(char *buf, size_t len, int client_id, long long seq) {
snprintf(buf, len, "rdma:%04d:%012lld", client_id, seq);
}
针对于一个worker的生命周期 #
static void * worker_main(void * arg) {
worker_config * worker = arg;
const test_config * cfg = worker - > cfg;
int state_count = worker - > last_client - worker - > first_client;
client_state * states;
int ret = 1;
states = calloc(state_count, sizeof( * states));
if (!states) {
fprintf(stderr, "thread %d failed to allocate client states\n",
worker - > thread_id);
return (void * )(long) 1;
}
for (int i = 0; i < state_count; i++) {
int client_id = worker - > first_client + i;
states[i].client_id = client_id;
states[i].requests = requests_for_client(cfg, client_id);
// 此处就尝试建立连接了
states[i].context = connect_rdma(cfg, client_id);
if (!states[i].context)
goto cleanup;
}
if (run_phase(states, state_count, cfg, 0) != 0)
goto cleanup;
for (int i = 0; i < state_count; i++)
states[i].processed = 0;
if (run_phase(states, state_count, cfg, 1) != 0)
goto cleanup;
printf("Valkey Over RDMA libvalkey thread[%d] clients %d-%d SET/GET %lld "
"requests [OK]\n",
worker - > thread_id, worker - > first_client, worker - > last_client - 1,
cfg - > requests);
ret = 0;
cleanup:
for (int i = 0; i < state_count; i++) {
if (states[i].context)
valkeyFree(states[i].context);
}
free(states);
return (void * )(long) ret;
}
差不多之后，我们需要把两个分支合并起来，在一个临时分支做个测试。
在CI中复现测试 #
因为这是单独的测试分支，相当于我们要在CI中复现出来错误，才能证明自己的正确性。
test:
sudo ./runtest-rdma --install-rxe
Valkey Over RDMA build test programs [OK]
Valkey Over RDMA test detect existing RDMA device [FAILED]
这里说没检测到设备。
但是在show-kernel-log中:
0s
Run sudo dmesg -c
[ 308.887784] rdma_rxe: loading out-of-tree module taints kernel.
[ 308.887792] rdma_rxe: module verification failed: signature and/or required key missing - tainting kernel
[ 308.890138] rdma_rxe: loaded
[ 308.896386] infiniband rxe_eth0: set active
[ 308.896390] infiniband rxe_eth0: added eth0
[ 309.206987] rdma_rxe: unloaded
我们可以看到设备应该是loaded过的，为什么？
这里总之是处理了一大堆py脚本的问题，看起来很难受。 我们最终解决了问题，然后使得终端报错的日志打印在CI页面上面，同时还正确的复现了问题，很好。
对于两个测试文件做一些抽象处理 #
须知重构抽象，乃软件本源。
先尝试做第一波抽象：
2.requests和minkeys maxkeys统一，这里的想法是两边都使用minkeys 和 maxkeys,然后默认选取合适的范围，使得此问题还是可以被触发 3.测试的KV pair 使用相同的逻辑进行填充()，当我们设置了KV之后，GET回来还需要进行校验()，保证逻辑对齐(),这里需要注意一下是什么意思。
需要修改的暂时主要为上面的三个点。
大型代码重构真的好难，完全不知道怎么下手，这里又是值得进行新学习的点。 就是各种逻辑全依赖在一起，你根本不知道开始动哪里，很难受。
分步骤一点点来吧 #
1.首先把公共结构体直接提出来，二者都使用这个结构体，如果存在对方没有的就先DEFAULT. 预期行为是命令行没有参数的时候，二者都使用自己的默认值（这个应该是CI中预期的行为），如果命令行指定了参数，二者使用相同的参数来进行测试。
比如这就是预期的参数：
┌──────────┬───────────────────────┬────────────────┐
│ 参数 │ rdma-test │ libvalkey-test │
├──────────┼───────────────────────┼────────────────┤
│ port │ 6379 │ 6379 │
├──────────┼───────────────────────┼────────────────┤
│ threads │ 0（单线程 main 模式） │ 16 │
├──────────┼───────────────────────┼────────────────┤
│ clients │ —（不用） │ 128 │
├──────────┼───────────────────────┼────────────────┤
│ pipeline │ — │ 384 │
├──────────┼───────────────────────┼────────────────┤
│ requests │ — │ 200000 │
├──────────┼───────────────────────┼────────────────┤
│ minkeys │ 128 │ — │
├──────────┼───────────────────────┼────────────────┤
│ maxkeys │ 8192 │ — │
├──────────┼───────────────────────┼────────────────┤
│ datasize │ 1024 │ 256 │
└──────────┴───────────────────────┴────────────────┘
没有什么问题。 目前两个测试共享了参数，具体的细节之后可以再修改。
2.接下来去掉request参数，把每个client的连接的request数量在minkeys和maxkeys内部做一个随机处理。 这个改动点应该不大，这个处理结束了。就是随机化一下，目前还是能稳定复现多线程的错误。
3.接下来是统一测试的逻辑,使用真随机，然后GET 回来时候进行compare的操作。