2. RDMA基础通信

任务目标

在两台配备RDMA网卡（如Mellanox ConnectX系列）的Linux节点间，通过ibverbs库实现一次RDMA Write操作：客户端将一段消息直接写入服务端预注册的内存缓冲区，无需服务端CPU参与数据拷贝。

环境依赖

• Linux系统，已安装rdma-core（包含libibverbs、librdmacm）及开发头文件
• 两台节点间RDMA链路可用（ibstatus确认状态为Active）
• 编译需链接：-libverbs

核心实现

以下代码实现完整闭环：服务端注册内存并等待写入，客户端发起RDMA Write后轮询完成队列确认。两端通过TCP交换QP信息（IP+端口），代码中已包含必要的握手逻辑。

/* rdma_write_demo.c — 编译: gcc -o rdma_write_demo rdma_write_demo.c -libverbs */
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <arpa/inet.h>
#include <sys/socket.h>
#include <infiniband/verbs.h>

#define BUFFER_SIZE 64
#define PORT 7471

struct conn_info {
    uint32_t addr;   /* 远端内存地址(网络序) */
    uint32_t rkey;   /* 远端RKey */
    uint32_t qp_num; /* 远端QP号 */
};

static int create_socket_conn(const char *ip, int is_server, struct conn_info *local, struct conn_info *remote) {
    int fd = socket(AF_INET, SOCK_STREAM, 0);
    struct sockaddr_in addr = {.sin_family = AF_INET, .sin_port = htons(PORT)};
    if (is_server) {
        addr.sin_addr.s_addr = INADDR_ANY;
        bind(fd, (struct sockaddr *)&addr, sizeof(addr));
        listen(fd, 1);
        int cfd = accept(fd, NULL, NULL);
        close(fd);
        fd = cfd;
    } else {
        inet_pton(AF_INET, ip, &addr.sin_addr);
        connect(fd, (struct sockaddr *)&addr, sizeof(addr));
    }
    /* 交换本地QP信息给远端，并接收远端QP信息 */
    send(fd, local, sizeof(*local), 0);
    recv(fd, remote, sizeof(*remote), 0);
    close(fd);
    return 0;
}

int main(int argc, char **argv) {
    if (argc < 2) { printf("用法: %s [server|client <server_ip>]\n", argv[0]); return 1; }
    int is_server = !strcmp(argv[1], "server");
    if (!is_server && argc < 3) { printf("客户端需指定服务端IP\n"); return 1; }

    /* 1. 打开RDMA设备 */
    struct ibv_device **devs = ibv_get_device_list(NULL);
    struct ibv_context *ctx = ibv_open_device(devs[0]);
    ibv_free_device_list(devs);
    struct ibv_pd *pd = ibv_alloc_pd(ctx);
    struct ibv_mr *mr;
    char *buf = malloc(BUFFER_SIZE);

    /* 2. 注册内存区域 */
    if (is_server) {
        mr = ibv_reg_mr(pd, buf, BUFFER_SIZE, IBV_ACCESS_LOCAL_WRITE);
        memset(buf, 0, BUFFER_SIZE);
    } else {
        mr = ibv_reg_mr(pd, buf, BUFFER_SIZE, 0);
        strcpy(buf, "Hello RDMA Write!");
    }

    /* 3. 创建QP */
    struct ibv_cq *cq = ibv_create_cq(ctx, 16, NULL, NULL, 0);
    struct ibv_qp_init_attr qp_attr = {
        .send_cq = cq, .recv_cq = cq,
        .qp_type = IBV_QPT_RC,
        .cap = {.max_send_wr = 4, .max_recv_wr = 4, .max_send_sge = 1, .max_recv_sge = 1}
    };
    struct ibv_qp *qp = ibv_create_qp(pd, &qp_attr);

    /* 4. 准备并交换QP信息 */
    struct conn_info local = {
        .addr = htonl((uintptr_t)buf),
        .rkey = htonl(mr->rkey),
        .qp_num = htonl(qp->qp_num)
    }, remote;
    create_socket_conn(is_server ? NULL : argv[2], is_server, &local, &remote);

    /* 5. 将QP切换到RTS状态 */
    struct ibv_qp_attr attr = {
        .qp_state = IBV_QPS_INIT,
        .pkey_index = 0,
        .port_num = 1,
        .qp_access_flags = IBV_ACCESS_LOCAL_WRITE
    };
    ibv_modify_qp(qp, &attr, IBV_QP_STATE | IBV_QP_PKEY_INDEX | IBV_QP_PORT | IBV_QP_ACCESS_FLAGS);
    attr.qp_state = IBV_QPS_RTR;
    attr.path_mtu = IBV_MTU_1024;
    attr.dest_qp_num = ntohl(remote.qp_num);
    attr.rq_psn = 0;
    attr.max_dest_rd_atomic = 1;
    attr.min_rnr_timer = 12;
    ibv_modify_qp(qp, &attr, IBV_QP_STATE | IBV_QP_MTU | IBV_QP_DEST_QPN | IBV_QP_RQ_PSN | IBV_QP_MAX_DEST_RD_ATOMIC | IBV_QP_MIN_RNR_TIMER);
    attr.qp_state = IBV_QPS_RTS;
    attr.timeout = 14;
    attr.retry_cnt = 7;
    attr.rnr_retry = 7;
    attr.sq_psn = 0;
    attr.max_rd_atomic = 1;
    ibv_modify_qp(qp, &attr, IBV_QP_STATE | IBV_QP_TIMEOUT | IBV_QP_RETRY_CNT | IBV_QP_RNR_RETRY | IBV_QP_SQ_PSN | IBV_QP_MAX_QP_RD_ATOMIC);

    /* 6. 客户端发起RDMA Write */
    if (!is_server) {
        struct ibv_sge sge = {.addr = (uintptr_t)buf, .length = BUFFER_SIZE, .lkey = 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 = ntohl(remote.addr), .rkey = ntohl(remote.rkey)}
        };
        struct ibv_send_wr *bad;
        ibv_post_send(qp, &wr, &bad);
    }

    /* 7. 等待完成事件 */
    struct ibv_wc wc;
    while (ibv_poll_cq(cq, 1, &wc) == 0) usleep(1000);
    if (wc.status != IBV_WC_SUCCESS) {
        printf("完成状态错误: %s\n", ibv_wc_status_str(wc.status));
        return 1;
    }

    /* 8. 输出结果 */
    if (is_server) {
        printf("服务端收到消息: %s\n", buf);
        printf("验证通过：RDMA Write数据已写入服务端内存\n");
    } else {
        printf("客户端发送完成，等待服务端确认...\n");
    }

    /* 清理资源 */
    ibv_destroy_qp(qp); ibv_destroy_cq(cq); ibv_dereg_mr(mr);
    ibv_dealloc_pd(pd); ibv_close_device(ctx); free(buf);
    return 0;
}

运行与验证

1. 编译：两端节点分别执行 gcc -o rdma_write_demo rdma_write_demo.c -libverbs
2. 启动服务端（节点A）：./rdma_write_demo server
3. 启动客户端（节点B）：./rdma_write_demo client <节点A的IP>
4. 判定通过条件：服务端控制台输出 服务端收到消息: Hello RDMA Write!，且客户端无错误退出。若输出乱码或超时，检查两端防火墙、RDMA链路状态（ibstatus）及QP状态是否成功切换至RTS。

说明：本示例使用TCP 7471端口仅用于交换QP元数据（内存地址、RKey、QP号），实际数据传输完全经RDMA链路完成。服务端在ibv_poll_cq阻塞等待写入完成事件，验证数据已直接落入其注册内存。