📚 课程目录第2章第一个RDMA程序

2. 第一个RDMA程序

任务目标

编写一个基于libibverbs的RDMA发送-接收程序，在InfiniBand或RoCE网络的两个节点间传输字符串消息。程序包含server端（被动接收）与client端（主动发送），使用可靠的连接（RC）服务类型，通过ibv_post_send和ibv_post_recv完成数据传输。

环境与依赖

Linux系统，已安装libibverbs-dev和rdma-core（或等价软件包）
两个节点间已配置好RDMA网络（InfiniBand或RoCE），可通过ibv_devinfo查看可用设备
编译命令：gcc -o rdma_send_recv rdma_send_recv.c -libverbs

核心实现

程序使用同步轮询完成队列（CQ）的方式等待操作完成，不依赖事件通道，以保持代码最小化。关键流程：client端建立连接后，注册内存区域（MR）、创建QP、发布接收请求（仅server端）、发送字符串、轮询CQ确认完成；server端先发布接收请求，等待接收完成后打印消息。

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <arpa/inet.h>
#include <infiniband/verbs.h>

#define BUFFER_SIZE 128
#define PORT_NUM 7471

struct conn_info {
    uint32_t qp_num;
    uint32_t rkey;
    uint64_t addr;
};

static int poll_cq(struct ibv_cq *cq, int num_completions) {
    struct ibv_wc wc;
    int completed = 0;
    while (completed < num_completions) {
        int ret = ibv_poll_cq(cq, 1, &wc);
        if (ret > 0) {
            if (wc.status != IBV_WC_SUCCESS) {
                fprintf(stderr, "CQ error: %s\n", ibv_wc_status_str(wc.status));
                return -1;
            }
            completed++;
        } else if (ret < 0) {
            perror("ibv_poll_cq");
            return -1;
        }
    }
    return 0;
}

static int setup_qp(struct ibv_pd *pd, struct ibv_cq *cq, struct ibv_qp **qp) {
    struct ibv_qp_init_attr attr = {
        .send_cq = cq,
        .recv_cq = cq,
        .qp_type = IBV_QPT_RC,
        .cap = { .max_send_wr = 1, .max_recv_wr = 1,
                 .max_send_sge = 1, .max_recv_sge = 1 }
    };
    *qp = ibv_create_qp(pd, &attr);
    if (!*qp) {
        perror("ibv_create_qp");
        return -1;
    }
    return 0;
}

static int exchange_info(int sockfd, struct conn_info *local, struct conn_info *remote) {
    if (send(sockfd, local, sizeof(*local), 0) != sizeof(*local)) return -1;
    if (recv(sockfd, remote, sizeof(*remote), 0) != sizeof(*remote)) return -1;
    return 0;
}

static int run_server(int sockfd) {
    struct ibv_device **dev_list;
    struct ibv_context *ctx;
    struct ibv_pd *pd;
    struct ibv_mr *mr;
    struct ibv_cq *cq;
    struct ibv_qp *qp;
    char *buf;
    struct conn_info local, remote;
    struct ibv_sge sge;
    struct ibv_recv_wr rr, *bad_rr;
    struct ibv_qp_attr attr;
    int ret = -1;

    dev_list = ibv_get_device_list(NULL);
    if (!dev_list) { perror("ibv_get_device_list"); return -1; }
    ctx = ibv_open_device(dev_list[0]);
    ibv_free_device_list(dev_list);
    if (!ctx) { perror("ibv_open_device"); return -1; }

    pd = ibv_alloc_pd(ctx);
    cq = ibv_create_cq(ctx, 2, NULL, NULL, 0);
    if (!pd || !cq) goto cleanup;
    if (setup_qp(pd, cq, &qp)) goto cleanup;

    buf = malloc(BUFFER_SIZE);
    if (!buf) goto cleanup;
    memset(buf, 0, BUFFER_SIZE);
    mr = ibv_reg_mr(pd, buf, BUFFER_SIZE, IBV_ACCESS_LOCAL_WRITE);
    if (!mr) { perror("ibv_reg_mr"); goto cleanup; }

    // 初始化QP到RTR状态
    memset(&attr, 0, sizeof(attr));
    attr.qp_state = IBV_QPS_INIT;
    attr.pkey_index = 0;
    attr.port_num = 1;
    attr.qp_access_flags = IBV_ACCESS_LOCAL_WRITE;
    if (ibv_modify_qp(qp, &attr, IBV_QP_STATE | IBV_QP_PKEY_INDEX |
                      IBV_QP_PORT | IBV_QP_ACCESS_FLAGS)) goto cleanup;

    local.qp_num = qp->qp_num;
    local.rkey = mr->rkey;
    local.addr = (uint64_t)buf;
    if (exchange_info(sockfd, &local, &remote)) goto cleanup;

    // RTR
    attr.qp_state = IBV_QPS_RTR;
    attr.path_mtu = IBV_MTU_1024;
    attr.dest_qp_num = remote.qp_num;
    attr.rq_psn = 0;
    attr.max_dest_rd_atomic = 1;
    attr.min_rnr_timer = 12;
    attr.ah_attr.is_global = 0;
    attr.ah_attr.port_num = 1;
    if (ibv_modify_qp(qp, &attr, IBV_QP_STATE | IBV_QP_PATH_MTU |
                      IBV_QP_DEST_QPN | IBV_QP_RQ_PSN |
                      IBV_QP_MAX_DEST_RD_ATOMIC | IBV_QP_MIN_RNR_TIMER |
                      IBV_QP_AV)) goto cleanup;

    // RTS
    attr.qp_state = IBV_QPS_RTS;
    attr.sq_psn = 0;
    attr.max_rd_atomic = 1;
    if (ibv_modify_qp(qp, &attr, IBV_QP_STATE | IBV_QP_SQ_PSN |
                      IBV_QP_MAX_ATOMIC)) goto cleanup;

    // 发布接收请求
    sge.addr = (uint64_t)buf;
    sge.length = BUFFER_SIZE;
    sge.lkey = mr->lkey;
    memset(&rr, 0, sizeof(rr));
    rr.sg_list = &sge;
    rr.num_sge = 1;
    if (ibv_post_recv(qp, &rr, &bad_rr)) { perror("ibv_post_recv"); goto cleanup; }

    // 等待接收完成
    if (poll_cq(cq, 1)) goto cleanup;
    printf("Server received: %s\n", buf);
    ret = 0;

cleanup:
    if (qp) ibv_destroy_qp(qp);
    if (mr) ibv_dereg_mr(mr);
    if (cq) ibv_destroy_cq(cq);
    if (pd) ibv_dealloc_pd(pd);
    if (ctx) ibv_close_device(ctx);
    free(buf);
    return ret;
}

static int run_client(int sockfd) {
    struct ibv_device **dev_list;
    struct ibv_context *ctx;
    struct ibv_pd *pd;
    struct ibv_mr *mr;
    struct ibv_cq *cq;
    struct ibv_qp *qp;
    char *buf;
    struct conn_info local, remote;
    struct ibv_sge sge;
    struct ibv_send_wr swr, *bad_swr;
    struct ibv_qp_attr attr;
    int ret = -1;

    dev_list = ibv_get_device_list(NULL);
    if (!dev_list) { perror("ibv_get_device_list"); return -1; }
    ctx = ibv_open_device(dev_list[0]);
    ibv_free_device_list(dev_list);
    if (!ctx) { perror("ibv_open_device"); return -1; }

    pd = ibv_alloc_pd(ctx);
    cq = ibv_create_cq(ctx, 2, NULL, NULL, 0);
    if (!pd || !cq) goto cleanup;
    if (setup_qp(pd, cq, &qp)) goto cleanup;

    buf = malloc(BUFFER_SIZE);
    if (!buf) goto cleanup;
    strcpy(buf, "Hello RDMA!");
    mr = ibv_reg_mr(pd, buf, BUFFER_SIZE, IBV_ACCESS_LOCAL_WRITE);
    if (!mr) { perror("ibv_reg_mr"); goto cleanup; }

    // 初始化QP到INIT
    memset(&attr, 0, sizeof(attr));
    attr.qp_state = IBV_QPS_INIT;
    attr.pkey_index = 0;
    attr.port_num = 1;
    attr.qp_access_flags = 0;
    if (ibv_modify_qp(qp, &attr, IBV_QP_STATE | IBV_QP_PKEY_INDEX |
                      IBV_QP_PORT | IBV_QP_ACCESS_FLAGS)) goto cleanup;

    local.qp_num = qp->qp_num;
    local.rkey = mr->rkey;
    local.addr = (uint64_t)buf;
    if (exchange_info(sockfd, &local, &remote)) goto cleanup;

    // RTR
    attr.qp_state = IBV_QPS_RTR;
    attr.path_mtu = IBV_MTU_1024;
    attr.dest_qp_num = remote.qp_num;
    attr.rq_psn = 0;
    attr.max_dest_rd_atomic = 1;
    attr.min_rnr_timer = 12;
    attr.ah_attr.is_global = 0;
    attr.ah_attr.port_num = 1;
    if (ibv_modify_qp(qp, &attr, IBV_QP_STATE | IBV_QP_PATH_MTU |
                      IBV_QP_DEST_QPN | IBV_QP_RQ_PSN |
                      IBV_QP_MAX_DEST_RD_ATOMIC | IBV_QP_MIN_RNR_TIMER |
                      IBV_QP_AV)) goto cleanup;

    // RTS
    attr.qp_state = IBV_QPS_RTS;
    attr.sq_psn = 0;
    attr.max_rd_atomic = 1;
    if (ibv_modify_qp(qp, &attr, IBV_QP_STATE | IBV_QP_SQ_PSN |
                      IBV_QP_MAX_ATOMIC)) goto cleanup;

    // 发送数据
    sge.addr = (uint64_t)buf;
    sge.length = strlen(buf) + 1;
    sge.lkey = mr->lkey;
    memset(&swr, 0, sizeof(swr));
    swr.sg_list = &sge;
    swr.num_sge = 1;
    swr.opcode = IBV_WR_SEND;
    swr.send_flags = IBV_SEND_SIGNALED;
    if (ibv_post_send(qp, &swr, &bad_swr)) { perror("ibv_post_send"); goto cleanup; }

    // 等待发送完成
    if (poll_cq(cq, 1)) goto cleanup;
    printf("Client sent: %s\n", buf);
    ret = 0;

cleanup:
    if (qp) ibv_destroy_qp(qp);
    if (mr) ibv_dereg_mr(mr);
    if (cq) ibv_destroy_cq(cq);
    if (pd) ibv_dealloc_pd(pd);
    if (ctx) ibv_close_device(ctx);
    free(buf);
    return ret;
}

int main(int argc, char *argv[]) {
    if (argc != 3) {
        fprintf(stderr, "Usage: %s <server|client> <server_ip>\n", argv[0]);
        return 1;
    }
    int sockfd;
    struct sockaddr_in addr;
    sockfd = socket(AF_INET, SOCK_STREAM, 0);
    if (sockfd < 0) { perror("socket"); return 1; }
    memset(&addr, 0, sizeof(addr));
    addr.sin_family = AF_INET;
    addr.sin_port = htons(PORT_NUM);

    if (strcmp(argv[1], "server") == 0) {
        int listen_fd = socket(AF_INET, SOCK_STREAM, 0);
        int opt = 1;
        setsockopt(listen_fd, SOL_SOCKET, SO_REUSEADDR, &opt, sizeof(opt));
        addr.sin_addr.s_addr = INADDR_ANY;
        bind(listen_fd, (struct sockaddr *)&addr, sizeof(addr));
        listen(listen_fd, 1);
        sockfd = accept(listen_fd, NULL, NULL);
        close(listen_fd);
        run_server(sockfd);
    } else {
        addr.sin_addr.s_addr = inet_addr(argv[2]);
        if (connect(sockfd, (struct sockaddr *)&addr, sizeof(addr))) {
            perror("connect");
            return 1;
        }
        run_client(sockfd);
    }
    close(sockfd);
    return 0;
}

运行与验证

编译：在两个节点上分别执行 gcc -o rdma_send_recv rdma_send_recv.c -libverbs
启动server（节点A）：./rdma_send_recv server
启动client（节点B）：./rdma_send_recv client <节点A的IP>
通过条件：server端打印 Server received: Hello RDMA!，client端打印 Client sent: Hello RDMA!，且两端程序均正常退出（退出码为0）。

预期输出（与代码逻辑严格一致）：client端发送缓冲区内容为"Hello RDMA!"，server端接收缓冲区打印相同字符串。若网络配置正确且QP状态迁移成功，两端各打印一行上述消息。

常见问题

ibv_open_device失败：确认当前节点有可用的RDMA设备（ibv_devinfo），且程序有权限访问/dev/infiniband/*。
QP迁移失败：检查两端QP状态机是否按INIT→RTR→RTS顺序执行，且交换的QPN、rkey、addr信息正确。
连接超时：确认TCP控制通道（端口7471）可达，且RDMA网络（IB或RoCE）在两端之间L2/L3连通。