Valkey Over RDMA: Multithread Tests (New Test Framework)

Table of Contents

I treat this as a new feature: use the libvalkey client to simulate load, with multithread support.

Test logic for this change
#

This is a new feature; we need a new PR.

Under tests/rdma, add libvalkey-test.c and use libvalkey’s client APIs for testing.

First study the existing test logic.

tests/rdma is not a conventional unit-test layout:

Directory layout:

File	Role
`run.py`	Main orchestrator: build, find RDMA NIC, start `valkey-server`, run `rdma-test`, cleanup
`rdma_env.py`	As root, setup/cleanup soft RDMA (RXE module, `rdma link add`)
`rdma-test.c`	C test client: hand-rolled RDMA CM/verbs, sends RESP
`Makefile`	Builds `rdma-test` (links `librdmacm`, `libibverbs`)

Roughly a smoke test — pizhenwei wrote a small RDMA client by hand.

But IO multithread is clearly not enabled.

Build once the same way. AI implemented it first; we then reviewed how:

make -j$(nproc) BUILD_RDMA=yes USE_FAST_FLOAT=yes
make -C tests/rdma

Broadly: use libvalkey client APIs to approximate this stress command, then adjust the build files. The code volume is large and needs real understanding.

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

We tested and changed a lot on the original branch with many commits, but I want this test alone in one PR: drop the partial testing from that branch so far, then replay on unstable to prove the bug.

You can tune parameters and reproduce yourself:

~/Project/valkey rdma-libvalkey-test* ❯ sudo ./runtest-rdma --install-rxe \
  --io-threads 8 \
  -c 500 -P 512 -n 1000000 --timeout 600

Under these conditions you will hit an assertion failure.

So parameters need a slight bump — but not too high, or runtime / OOM becomes a problem. Overall some knobs went up.

What does `libvalkey-test.c` do?
#

Think of it as: use the official libvalkey client to reproduce valkey-benchmark-style pipeline stress over RDMA.

What is this test verifying?
#

Compared with rdma-test.c in the same directory:

Capability / file	`rdma-test`	`libvalkey-test`
Client stack	Hand-written RDMA CM / CQ / QP	libvalkey (`deps/libvalkey`)
Protocol	Hand-built RESP, hand-read RDMA buffers	libvalkey formats RESP + RDMA transport
Concurrency	4 threads, 1 connection each	16 workers × multiple connections per worker
Mode	One-by-one SET/GET	Pipeline: append a batch, flush, drain
Goal	Functional smoke (PING/SET/GET/BGSAVE)	Stability under RDMA + IO threads + high pipeline

The bug you are fixing (cmd_queue.len == 0 assert) needs both:

Many connections with inflight pipelined commands
Server IO threads on the read path (run.py enables this only in the second phase)

This program is the client-side load generator designed for that.

Overall architecture
#

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

Basic idea: many threads, each managing many connections.

Note the thread vs connection split:

  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;
      }
  }

Distribute total request count:

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;
}

When not divisible, the first requests % clients clients get one extra, so the sum equals cfg->requests exactly.

Core data structures
#

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;

Field	Meaning
`test_config`	Global params; all workers read-only
`worker_config`	Client range for this pthread `[first_client, last_client)`
`client_state.processed`	Completed SET (or GET) count on this connection
`client_state.pending`	Commands appended in the current pipeline batch but not yet drained
`client_state.context`	libvalkey connection; has `obuf` (write buffer) and `reader` (parse)

Also use distinct keys per connection to avoid contention:

static void make_key(char *buf, size_t len, int client_id, long long seq) {
  snprintf(buf, len, "rdma:%04d:%012lld", client_id, seq);
}

Lifecycle of one 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;
}

After that, merge the two branches and test on a temporary branch.

Reproducing in CI
#

Because this is a standalone test branch, we need CI to reproduce the failure to prove correctness.

test:

sudo ./runtest-rdma --install-rxe
  
Valkey Over RDMA build test programs [OK]
Valkey Over RDMA test detect existing RDMA device [FAILED]

It says no device detected.

But in 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

We can see the device was loaded — so why?

Ultimately a pile of Python script issues; painful, but we fixed them so terminal error logs show on the CI page, and we correctly reproduced the bug. Good.

Abstracting the two test files
#

Reminder: refactoring / abstraction is software’s root craft.

First wave of abstraction:

Unify requests with minkeys/maxkeys — both sides use minkeys/maxkeys with sensible defaults so the bug can still be triggered
Fill KV pairs with the same logic; after SET, GET must validate (compare) so both tests align. Pay attention to what that means.

Those three points are the main temporary targets.

Large refactors are hard — you don’t know where to start. Another learning point: everything is tangled; moving any piece hurts.

Step by step
#

First extract shared structs; both tests use them; fields one side lacks get DEFAULT. Expected: with no CLI args, each test keeps its own defaults (CI behavior); with CLI args, both share the same parameters.

Example expected defaults:

  ┌──────────┬───────────────────────┬────────────────┐
  │   参数   │       rdma-test       │ libvalkey-test │
  ├──────────┼───────────────────────┼────────────────┤
  │ port     │ 6379                  │ 6379           │
  ├──────────┼───────────────────────┼────────────────┤
  │ threads  │ 0（单线程 main 模式） │ 16             │
  ├──────────┼───────────────────────┼────────────────┤
  │ clients  │ —（不用）             │ 128            │
  ├──────────┼───────────────────────┼────────────────┤
  │ pipeline │ —                     │ 384            │
  ├──────────┼───────────────────────┼────────────────┤
  │ requests │ —                     │ 200000         │
  ├──────────┼───────────────────────┼────────────────┤
  │ minkeys  │ 128                   │ —              │
  ├──────────┼───────────────────────┼────────────────┤
  │ maxkeys  │ 8192                  │ —              │
  ├──────────┼───────────────────────┼────────────────┤
  │ datasize │ 1024                  │ 256            │
  └──────────┴───────────────────────┴────────────────┘

Looks fine. The two tests now share parameters; details can change later.

Next drop the requests parameter; randomize each client’s request count within minkeys/maxkeys. Small change; done. Still stably reproduces the multithread bug.
Next unify test logic: true random keys/values, then compare on GET.

Test logic for this change #

What does libvalkey-test.c do? #

What is this test verifying? #

Overall architecture #

Core data structures #

Lifecycle of one worker #

Reproducing in CI #

Abstracting the two test files #

Step by step #