3. RDMA写操作实现

任务目标

实现一个RDMA写操作（ibv_post_send + IBV_WR_RDMA_WRITE），将本地缓冲区数据直接写入远程预注册内存。远程节点CPU不感知数据到达，仅由网卡完成内存写入。本案例聚焦单边写操作的最小可运行闭环。

环境与依赖

  • Linux系统，已安装rdma-core开发库（含libibverbs、librdmacm）
  • 支持RDMA的网卡（如Mellanox CX系列）或软RoCE环境（rdma_rxe模块）
  • 编译依赖：libibverbs-dev、librdmacm-dev

核心实现

以下代码实现完整的RDMA写操作闭环：客户端（主动方）向服务端（被动方）发起写操作。服务端仅注册内存并等待，不轮询接收数据；客户端通过rdma_cm建立连接后，使用ibv_post_send提交RDMA写请求。

// rdma_write.c — 编译: gcc -o rdma_write rdma_write.c -lrdmacm -libverbs
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <rdma/rdma_cma.h>
#include <infiniband/verbs.h>
#include <arpa/inet.h>

#define BUFFER_SIZE 64
#define PORT 7471

struct conn_info {
    struct rdma_cm_id *id;
    struct ibv_mr *mr;
    char *buf;
};

// 注册内存区域
static int setup_qp_and_mr(struct rdma_cm_id *id, struct conn_info *c) {
    c->buf = calloc(1, BUFFER_SIZE);
    if (!c->buf) return -1;

    struct ibv_pd *pd = ibv_alloc_pd(id->verbs);
    if (!pd) { perror("alloc_pd"); return -1; }

    c->mr = ibv_reg_mr(pd, c->buf, BUFFER_SIZE,
                       IBV_ACCESS_LOCAL_WRITE | IBV_ACCESS_REMOTE_WRITE);
    if (!c->mr) { perror("reg_mr"); return -1; }

    struct ibv_qp_init_attr attr = {
        .qp_type = IBV_QPT_RC,
        .send_cq = ibv_create_cq(id->verbs, 16, NULL, NULL, 0),
        .recv_cq = ibv_create_cq(id->verbs, 16, NULL, NULL, 0),
        .cap = { .max_send_wr = 4, .max_recv_wr = 4,
                 .max_send_sge = 1, .max_recv_sge = 1 }
    };
    if (rdma_create_qp(id, pd, &attr)) { perror("create_qp"); return -1; }
    return 0;
}

// 交换内存地址和rkey
static int exchange_info(struct rdma_cm_id *id, struct conn_info *c,
                         uint64_t *remote_addr, uint32_t *remote_rkey) {
    struct msg { uint64_t addr; uint32_t rkey; } local, remote;
    local.addr = (uint64_t)c->buf;
    local.rkey = c->mr->rkey;

    if (rdma_post_recv(id, NULL, &remote, sizeof(remote),
                       (struct ibv_sge[]){{ .addr = (uint64_t)&remote,
                       .length = sizeof(remote), .lkey = c->mr->lkey }}) ||
        rdma_post_send(id, NULL, &local, sizeof(local),
                       (struct ibv_sge[]){{ .addr = (uint64_t)&local,
                       .length = sizeof(local), .lkey = c->mr->lkey }}, 0)) {
        perror("exchange"); return -1;
    }
    // 等待完成（简化：轮询）
    struct ibv_cq *cq; struct ibv_wc wc;
    while (ibv_poll_cq(id->send_cq, 1, &wc) == 0);
    while (ibv_poll_cq(id->recv_cq, 1, &wc) == 0);

    *remote_addr = remote.addr;
    *remote_rkey = remote.rkey;
    return 0;
}

// 执行RDMA写操作
static int do_rdma_write(struct rdma_cm_id *id, struct conn_info *c,
                         uint64_t remote_addr, uint32_t remote_rkey) {
    strcpy(c->buf, "Hello RDMA Write!");  // 待写入数据

    struct ibv_sge sge = { .addr = (uint64_t)c->buf,
                           .length = BUFFER_SIZE, .lkey = c->mr->lkey };
    struct ibv_send_wr wr = {
        .opcode = IBV_WR_RDMA_WRITE,
        .send_flags = IBV_SEND_SIGNALED,
        .sg_list = &sge, .num_sge = 1,
        .wr.rdma = { .remote_addr = remote_addr, .rkey = remote_rkey }
    };
    struct ibv_send_wr *bad_wr;
    if (ibv_post_send(id->qp, &wr, &bad_wr)) { perror("post_send"); return -1; }

    struct ibv_cq *cq; struct ibv_wc wc;
    while (ibv_poll_cq(id->send_cq, 1, &wc) == 0);  // 等待写完成
    if (wc.status != IBV_WC_SUCCESS) {
        fprintf(stderr, "WC status: %s\n", ibv_wc_status_str(wc.status));
        return -1;
    }
    return 0;
}

int main(int argc, char **argv) {
    if (argc != 3) {
        printf("用法: %s <server|client> <server_ip>\n", argv[0]);
        return 1;
    }
    int is_server = strcmp(argv[1], "server") == 0;
    struct rdma_cm_id *listen_id = NULL, *id = NULL;
    struct rdma_event_channel *ec = rdma_create_event_channel();
    struct conn_info conn = {0};

    if (is_server) {
        rdma_create_id(ec, &listen_id, NULL, RDMA_PS_TCP);
        struct sockaddr_in addr = { .sin_family = AF_INET, .sin_port = htons(PORT) };
        rdma_bind_addr(listen_id, (struct sockaddr *)&addr);
        rdma_listen(listen_id, 1);
        rdma_get_request(listen_id, &id);
        setup_qp_and_mr(id, &conn);
        uint64_t raddr; uint32_t rrkey;
        exchange_info(id, &conn, &raddr, &rrkey);
        printf("服务端: 等待客户端写入...\n");
        sleep(2);  // 给客户端时间执行写操作
        printf("服务端缓冲区内容: %s\n", conn.buf);
    } else {
        rdma_create_id(ec, &id, NULL, RDMA_PS_TCP);
        struct sockaddr_in addr = { .sin_family = AF_INET, .sin_port = htons(PORT) };
        inet_pton(AF_INET, argv[2], &addr.sin_addr);
        rdma_resolve_addr(id, NULL, (struct sockaddr *)&addr, 2000);
        rdma_resolve_route(id, 2000);
        rdma_connect(id, NULL);
        setup_qp_and_mr(id, &conn);
        uint64_t raddr; uint32_t rrkey;
        exchange_info(id, &conn, &raddr, &rrkey);
        do_rdma_write(id, &conn, raddr, rrkey);
        printf("客户端: RDMA写完成\n");
    }
    return 0;
}

运行与验证

  1. 编译：gcc -o rdma_write rdma_write.c -lrdmacm -libverbs
  2. 启动服务端（节点A）：./rdma_write server
  3. 启动客户端（节点B）：./rdma_write client <服务端IP>
  4. 验证标准：服务端控制台打印的缓冲区内容为Hello RDMA Write!，且服务端CPU未参与数据接收（无recv完成事件处理）。客户端打印“RDMA写完成”且无错误输出。

关键说明

  • 服务端必须注册内存时带IBV_ACCESS_REMOTE_WRITE权限，否则客户端写操作会返回IBV_WC_REM_ACCESS_ERR。
  • 本示例通过rdma_post_recv/send交换内存地址和rkey，这是建立连接后必需的握手步骤；实际数据写入不经过接收队列。
  • 若在单机验证，需加载软RoCE：modprobe rdma_rxe并配置rxe网卡，或使用支持loopback的硬件网卡。
  • 服务端sleep(2)仅为演示等待写入完成；生产环境应使用事件通知或轮询远程内存值确认。
← 上一章 📖 返回目录 下一章 →