【RDMA项目】如何使用rdma-core进行调用开发一个实战项目

RDMA (Remote Direct Memory Access) 是一种网络协议，可以在计算节点之间实现高效的内存数据传输，而无需CPU的干预。rdma-core 是 RDMA 的一个用户空间库，提供了一些简单易用的接口来使用 RDMA 功能。

开发了一套高级 RDMA（远程直接内存访问）连接和数据传输系统，使用 rdma-core 库（包括 rdma_cm 和 ibverbs）编写，适用于高性能计算和实时数据处理。该项目分为服务器和客户端两部分，能够在实际网络环境中实现 RDMA 连接和数据传输。通过零拷贝技术和高效的网络编程，系统显著提高了数据传输速度和减少了延迟。项目成功验证了 RDMA 在高性能和低延迟场景中的应用价值，为需要高效数据传输的应用场景提供了可靠的解决方案。

在这个项目中，rdma_cm（RDMA Connection Manager）用于管理RDMA连接的创建、绑定、监听、接受和断开。rdma_cm 提供了一套API，使得RDMA编程变得更为简便，抽象了底层复杂的连接管理过程。

在这个项目中，ibverbs（Infiniband Verbs API）用于管理低级别的Infiniband操作，如保护域、内存区域、完成队列和队列对（QP）的分配和管理。

基础知识参考博文：

一文带你了解什么是RDMA

RDMA 高性能架构基本原理与设计方案

RDMA之RoCE & Soft-RoCE_software roce

如何使用rdma-core来实现RDMA操作

教你在虚拟机上把普通网卡配置成softroce设备来运行rdma-core中的示例程序-CSDN博客

目录：

一、环境准备：

1.1 安装依赖

1.2 安装 rdma-core

1.3 支持RDMA的机器

二、使用 RDMA（主要用于测试RDMA 环境配置是否正确）

2.1 编写 RDMA 程序

2.2 编译和运行程序

三、进一步的 RDMA-core 操作示例

3.1 服务器代码

3.1.1 初始化和准备工作

3.1.2 设置和监听

3.1.3 接受连接

3.1.4 资源分配和QP创建

3.1.5 内存注册和接收操作

3.1.6 轮询和处理完成队列

3.1.7 清理资源

3.1.8 主函数

3.2 客户端代码

3.2.1 初始化和准备工作

3.2.2 创建和解析连接

3.2.3 路由解析

3.2.4 资源分配和QP创建

3.2.5 内存注册与连接

3.2.6 数据发送

3.2.7 清理资源

3.2.8 主函数：

3.3 编译和运行

3.5 结果

四、rdma-core 实现更复杂的 RDMA 连接和数据传输操作

4.1 RDMA 操作基础

4.2 设备发现与上下文创建

4.1.1 变量定义

4.1.2 获取设备列表：

4.1.3 选择设备：

4.1.4 获取设备上下文

4.1.5 打印设备名称并且清理资源

4.1.6 编译运行：

4.1.7 示例输出：

4.3 内存注册与队列对创建

4.3.1 获取InfiniBand设备列表并打开设备:（此过程就是4.2介绍的过程）

4.3.2 创建保护域（Protection Domain，PD）:

4.3.3 注册内存区域:

4.3.4 创建完成队列（Completion Queue，CQ）:

4.3.5 设置队列对属性并创建队列对:

4.3.6 清理资源:

4.3.7 演示效果

4.4 建立 RDMA 连接

1. 服务器端代码（与3.1代码略有不同）

2. 客户端代码（与3.2代码相同）

4.5 编译和运行

4.6 结果

五、验证RDMA 在高性能和低延迟场景中的应用价值

5.1 性能测试方法

5.1.2 延迟测试：

5.2.3 零拷贝效应：

5.2.4 资源使用率分析：

5.2 实验设计

5.2.1 测试环境搭建：

5.2.2 基准测试：

5.2.3 压力测试：

5.3 测试指标

5.3.1 吞吐量（Throughput）：

5.3.2 延迟（Latency）：

5.3.3 资源使用率（Resource Utilization）：

5.4 实验结果分析

5.5 实验示例

一、环境准备：

1.1 安装依赖

在安装 rdma-core 之前，确保你的系统已经安装了相关依赖库。对于基于 Debian 的系统，可以使用以下命令：

sudo apt-get update
sudo apt-get install build-essential cmake libnuma-dev libibverbs-dev

对于基于 Red Hat 的系统，可以使用以下命令：

sudo yum groupinstall "Development Tools"
sudo yum install cmake libibverbs-devel numactl-devel

1.2 安装 rdma-core

你可以从源码编译并安装 rdma-core。首先克隆 rdma-core 的 GitHub 仓库：

git clone https://github.com/linux-rdma/rdma-core.git
cd rdma-core

然后使用 CMake 进行编译和安装：

mkdir build
cd build
cmake ..
make
sudo make install

1.3 支持RDMA的机器

准备俩台支持RDMA的设备机器进行测试调试

二、使用 RDMA（主要用于测试RDMA 环境配置是否正确）

2.1 编写 RDMA 程序

rdma-core 提供了多个库，例如 libibverbs 和 librdmacm，用于不同的 RDMA 操作。以下是一个简单的例子，展示了如何使用 libibverbs 库进行基本的 RDMA 操作。

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

int main() {
    struct ibv_device **dev_list;  // 用于存储设备列表的指针数组
    struct ibv_device *ib_dev;     // 用于存储选择的设备指针
    struct ibv_context *ctx;       // 用于存储设备上下文的指针

    // 获取设备列表
    dev_list = ibv_get_device_list(NULL);
    if (!dev_list) {  // 检查设备列表是否获取成功
        perror("Failed to get devices list");
        return EXIT_FAILURE;
    }

    // 选择第一个设备
    ib_dev = dev_list[0];
    if (!ib_dev) {  // 检查是否找到至少一个设备
        fprintf(stderr, "No IB devices found\n");
        ibv_free_device_list(dev_list);  // 释放设备列表
        return EXIT_FAILURE;
    }

    // 获取设备的上下文
    ctx = ibv_open_device(ib_dev);
    if (!ctx) {  // 检查是否成功打开设备上下文
        perror("Failed to open device");
        ibv_free_device_list(dev_list);  // 释放设备列表
        return EXIT_FAILURE;
    }

    printf("Device %s opened\n", ib_dev->name);  // 输出设备名称表示成功打开

    // 清理
    ibv_close_device(ctx);  // 关闭设备上下文
    ibv_free_device_list(dev_list);  // 释放设备列表

    return EXIT_SUCCESS;  // 程序成功执行完毕
}

2.2 编译和运行程序

将上述代码保存到一个文件中，例如 rdma_example.c，然后使用以下命令进行编译和运行：

gcc -o rdma_example rdma_example.c -libverbs
./rdma_example

恭喜你成功打开了 RDMA 设备 rxe0！这说明你的 RDMA 环境已经配置正确，能够正常工作。接下来，可以尝试更多复杂的 RDMA 操作，如建立连接、数据传输等。以下是一个更复杂的示例，展示如何进行基本的 RDMA 数据传输。

三、进一步的 RDMA-core 操作示例

我们将展示一个简单的 RDMA 客户端-服务器程序，其中客户端向服务器发送数据，服务器接收数据。这只是一个简单的示例，实际应用中可能需要更多的错误处理和资源管理。

3.1 服务器代码

保存以下代码为 rdma_server.c：

实现了一种使用RDMA（远程直接数据存取）技术的简单服务器程序。RDMA允许在计算机之间进行高速数据传输，而无需操作系统干预。

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <rdma/rdma_cma.h>
#include <infiniband/verbs.h>

#define BUFFER_SIZE 1024

void run_server() {
    struct rdma_event_channel *ec = rdma_create_event_channel(); // 创建RDMA事件通道
    struct rdma_cm_id *listener = NULL, *conn = NULL; // 定义监听和连接ID
    struct rdma_addrinfo hints, *res; // 定义地址信息结构体
    struct ibv_pd *pd = NULL; // 定义保护域
    struct ibv_mr *mr = NULL; // 定义内存注册区域
    struct ibv_comp_channel *comp_chan = NULL; // 定义完成通道
    struct ibv_cq *cq = NULL; // 定义完成队列
    struct ibv_qp_init_attr qp_attr; // 定义QP初始化属性
    char buf[BUFFER_SIZE]; // 定义缓冲区
    int ret; // 定义返回值变量

    memset(&hints, 0, sizeof(hints)); // 将hints结构体清零
    hints.ai_flags = RAI_PASSIVE; // 设置hints为被动模式
    hints.ai_port_space = RDMA_PS_TCP; // 设置地址空间为TCP

    ret = rdma_getaddrinfo(NULL, "20079", &hints, &res); // 获取地址信息
    if (ret) {
        perror("rdma_getaddrinfo"); // 如果失败，输出错误信息并退出
        exit(EXIT_FAILURE);
    }

    ret = rdma_create_id(ec, &listener, NULL, RDMA_PS_TCP); // 创建RDMA标识符
    if (ret) {
        perror("rdma_create_id"); // 如果失败，输出错误信息并退出
        exit(EXIT_FAILURE);
    }

    ret = rdma_bind_addr(listener, res->ai_src_addr); // 绑定地址
    if (ret) {
        perror("rdma_bind_addr"); // 如果失败，输出错误信息并退出
        exit(EXIT_FAILURE);
    }

    ret = rdma_listen(listener, 0); // 开始监听
    if (ret) {
        perror("rdma_listen"); // 如果失败，输出错误信息并退出
        exit(EXIT_FAILURE);
    }

    printf("Server is listening on port 20079...\n"); // 打印服务器正在监听的消息

    struct rdma_cm_event *event; // 定义RDMA事件
    ret = rdma_get_cm_event(ec, &event); // 获取一个连接管理事件
    if (ret) {
        perror("rdma_get_cm_event"); // 如果失败，输出错误信息并退出
        exit(EXIT_FAILURE);
    }

    if (event->event == RDMA_CM_EVENT_CONNECT_REQUEST) { // 检查事件类型是否为连接请求
        conn = event->id; // 获取连接ID
        rdma_ack_cm_event(event); // 确认事件
    } else {
        fprintf(stderr, "Unexpected event: %d\n", event->event); // 如果事件类型不是预期的，输出错误信息并退出
        rdma_ack_cm_event(event); // 确认事件
        exit(EXIT_FAILURE);
    }

    pd = ibv_alloc_pd(conn->verbs); // 分配保护域
    if (!pd) {
        perror("ibv_alloc_pd"); // 如果失败，输出错误信息并退出
        exit(EXIT_FAILURE);
    }

    comp_chan = ibv_create_comp_channel(conn->verbs); // 创建完成通道
    if (!comp_chan) {
        perror("ibv_create_comp_channel"); // 如果失败，输出错误信息并退出
        exit(EXIT_FAILURE);
    }

    cq = ibv_create_cq(conn->verbs, 10, NULL, comp_chan, 0); // 创建完成队列
    if (!cq) {
        perror("ibv_create_cq"); // 如果失败，输出错误信息并退出
        exit(EXIT_FAILURE);
    }

    memset(&qp_attr, 0, sizeof(qp_attr)); // 清零QP初始化属性结构体
    qp_attr.cap.max_send_wr = 1; // 设置最大发送工作请求数
    qp_attr.cap.max_recv_wr = 1; // 设置最大接收工作请求数
    qp_attr.cap.max_send_sge = 1; // 设置最大发送SGE数
    qp_attr.cap.max_recv_sge = 1; // 设置最大接收SGE数
    qp_attr.send_cq = cq; // 关联发送完成队列
    qp_attr.recv_cq = cq; // 关联接收完成队列
    qp_attr.qp_type = IBV_QPT_RC; // 设置QP类型为RC

    ret = rdma_create_qp(conn, pd, &qp_attr); // 创建QP
    if (ret) {
        perror("rdma_create_qp"); // 如果失败，输出错误信息并退出
        exit(EXIT_FAILURE);
    }

    mr = ibv_reg_mr(pd, buf, BUFFER_SIZE, IBV_ACCESS_LOCAL_WRITE | IBV_ACCESS_REMOTE_WRITE); // 注册内存区域
    if (!mr) {
        perror("ibv_reg_mr"); // 如果失败，输出错误信息并退出
        exit(EXIT_FAILURE);
    }

    struct ibv_recv_wr wr, *bad_wr = NULL; // 定义接收工作请求和错误工作请求指针
    struct ibv_sge sge; // 定义SGE

    sge.addr = (uintptr_t)buf; // 设置SGE地址为缓冲区地址
    sge.length = BUFFER_SIZE; // 设置SGE长度为缓冲区大小
    sge.lkey = mr->lkey; // 设置SGE的本地密钥为内存区域的密钥

    wr.wr_id = 0; // 设置工作请求ID为0
    wr.next = NULL; // 设置下一个工作请求为空
    wr.sg_list = &sge; // 设置工作请求的SGE列表
    wr.num_sge = 1; // 设置SGE数量为1

    ret = ibv_post_recv(conn->qp, &wr, &bad_wr); // 发起接收工作请求
    if (ret) {
        perror("ibv_post_recv"); // 如果失败，输出错误信息并退出
        exit(EXIT_FAILURE);
    }

    struct rdma_conn_param cm_params; // 定义连接参数
    memset(&cm_params, 0, sizeof(cm_params)); // 清零连接参数结构体
    ret = rdma_accept(conn, &cm_params); // 接受连接请求
    if (ret) {
        perror("rdma_accept"); // 如果失败，输出错误信息并退出
        exit(EXIT_FAILURE);
    }

    struct ibv_wc wc; // 定义工作完成结构体
    while ((ret = ibv_poll_cq(cq, 1, &wc)) == 0); // 轮询完成队列直到有工作完成
    if (ret < 0) {
        perror("ibv_poll_cq"); // 如果失败，输出错误信息并退出
        exit(EXIT_FAILURE);
    }

    if (wc.status != IBV_WC_SUCCESS) { // 检查工作完成状态是否成功
        fprintf(stderr, "Failed status %s (%d) for wr_id %d\n",
                ibv_wc_status_str(wc.status), wc.status, (int) wc.wr_id); // 如果失败，输出错误信息并退出
        exit(EXIT_FAILURE);
    }

    printf("Received message: %s\n", buf); // 打印接收到的消息

    rdma_disconnect(conn); // 断开连接
    rdma_destroy_qp(conn); // 销毁QP
    ibv_dereg_mr(mr); // 解注册内存区域
    ibv_destroy_cq(cq); // 销毁完成队列
    ibv_destroy_comp_channel(comp_chan); // 销毁完成通道
    ibv_dealloc_pd(pd); // 释放保护域
    rdma_destroy_id(conn); // 销毁连接ID
    rdma_destroy_id(listener); // 销毁监听ID
    rdma_destroy_event_channel(ec); // 销毁事件通道
    rdma_freeaddrinfo(res); // 释放地址信息
}

int main() {
    run_server(); // 运行服务器
    return 0; // 返回0表示正常退出
}

3.1.1 初始化和准备工作

1. 创建并初始化RDMA事件通道：

struct rdma_event_channel *ec = rdma_create_event_channel(); // 创建RDMA事件通道

2. 定义变量：

    struct rdma_cm_id *listener = NULL, *conn = NULL; // 定义监听和连接ID
    struct rdma_addrinfo hints, *res; // 定义地址信息结构体
    struct ibv_pd *pd = NULL; // 定义保护域
    struct ibv_mr *mr = NULL; // 定义内存注册区域
    struct ibv_comp_channel *comp_chan = NULL; // 定义完成通道
    struct ibv_cq *cq = NULL; // 定义完成队列
    struct ibv_qp_init_attr qp_attr; // 定义QP初始化属性
    char buf[BUFFER_SIZE]; // 定义缓冲区
    int ret; // 定义返回值变量

listener和conn：用于监听和连接的RDMA标识符。
hints和res：用于存储地址信息。
pd、mr、comp_chan和cq：分别代表保护域、内存注册区、完成通道和完成队列。
qp_attr：用于配置QP的初始化属性。
buf：用于存储接收到的数据。
ret：用于存储各个操作的返回值。

3. 配置地址信息：

    memset(&hints, 0, sizeof(hints)); // 将hints结构体清零
    hints.ai_flags = RAI_PASSIVE; // 设置hints为被动模式
    hints.ai_port_space = RDMA_PS_TCP; // 设置地址空间为TCP

4. 获取RDMA地址：

    ret = rdma_getaddrinfo(NULL, "20079", &hints, &res); // 获取地址信息

3.1.2 设置和监听

1. 创建RDMA标识符：

    ret = rdma_create_id(ec, &listener, NULL, RDMA_PS_TCP); // 创建RDMA标识符

2. 绑定地址并开始监听：

    ret = rdma_bind_addr(listener, res->ai_src_addr);//绑定地址
    ret = rdma_listen(listener, 0);//开始监听

3.1.3 接受连接

1. 等待并接受连接请求：

    struct rdma_cm_event *event;//定义RDMA事件
    ret = rdma_get_cm_event(ec, &event);//获取一个连接管理事件

2.. 处理连接请求：

    if (event->event == RDMA_CM_EVENT_CONNECT_REQUEST) { // 检查事件类型是否为连接请求
        conn = event->id; // 获取连接ID
        rdma_ack_cm_event(event); // 确认事件
    } else {
        fprintf(stderr, "Unexpected event: %d\n", event->event); // 如果事件类型不是预期的，输出错误信息并退出
        rdma_ack_cm_event(event); // 确认事件
        exit(EXIT_FAILURE);
    }

3.1.4 资源分配和QP创建

1. 分配保护域、创建完成通道和完成队列：

    pd = ibv_alloc_pd(conn->verbs); // 分配保护域
    comp_chan = ibv_create_comp_channel(conn->verbs); // 创建完成通道
    cq = ibv_create_cq(conn->verbs, 10, NULL, comp_chan, 0); // 创建完成队列

2. 配置并创建QP：

    memset(&qp_attr, 0, sizeof(qp_attr)); // 清零QP初始化属性结构体
    qp_attr.cap.max_send_wr = 1; // 设置最大发送工作请求数
    qp_attr.cap.max_recv_wr = 1; // 设置最大接收工作请求数
    qp_attr.cap.max_send_sge = 1; // 设置最大发送SGE数
    qp_attr.cap.max_recv_sge = 1; // 设置最大接收SGE数
    qp_attr.send_cq = cq; // 关联发送完成队列
    qp_attr.recv_cq = cq; // 关联接收完成队列
    qp_attr.qp_type = IBV_QPT_RC; // 设置QP类型为RC

    ret = rdma_create_qp(conn, pd, &qp_attr); // 创建QP

3.1.5 内存注册和接收操作

1. 注册内存区域：

    mr = ibv_reg_mr(pd, buf, BUFFER_SIZE, IBV_ACCESS_LOCAL_WRITE | IBV_ACCESS_REMOTE_WRITE);//注册内存区域

2. 准备接收工作请求：

    struct ibv_recv_wr wr, *bad_wr = NULL; // 定义接收工作请求和错误工作请求指针
    struct ibv_sge sge; // 定义SGE

    sge.addr = (uintptr_t)buf; // 设置SGE地址为缓冲区地址
    sge.length = BUFFER_SIZE; // 设置SGE长度为缓冲区大小
    sge.lkey = mr->lkey; // 设置SGE的本地密钥为内存区域的密钥

    wr.wr_id = 0; // 设置工作请求ID为0
    wr.next = NULL; // 设置下一个工作请求为空
    wr.sg_list = &sge; // 设置工作请求的SGE列表
    wr.num_sge = 1; // 设置SGE数量为1

    ret = ibv_post_recv(conn->qp, &wr, &bad_wr); // 发起接收工作请求

3. 接受连接：

    struct rdma_conn_param cm_params; // 定义连接参数
    memset(&cm_params, 0, sizeof(cm_params)); // 清零连接参数结构体
    ret = rdma_accept(conn, &cm_params); // 接受连接请求

3.1.6 轮询和处理完成队列

1. 轮询完成队列，等待数据接收：

    struct ibv_wc wc; // 定义工作完成结构体
    while ((ret = ibv_poll_cq(cq, 1, &wc)) == 0); // 轮询完成队列直到有工作完成

2. 检查完成状态并处理数据：

    if (wc.status != IBV_WC_SUCCESS) { // 检查工作完成状态是否成功
        fprintf(stderr, "Failed status %s (%d) for wr_id %d\n",
                ibv_wc_status_str(wc.status), wc.status, (int) wc.wr_id); // 如果失败，输出错误信息并退出
        exit(EXIT_FAILURE);
    }

3.1.7 清理资源

断开连接并释放资源：

    rdma_disconnect(conn); // 断开连接
    rdma_destroy_qp(conn); // 销毁QP
    ibv_dereg_mr(mr); // 解注册内存区域
    ibv_destroy_cq(cq); // 销毁完成队列
    ibv_destroy_comp_channel(comp_chan); // 销毁完成通道
    ibv_dealloc_pd(pd); // 释放保护域
    rdma_destroy_id(conn); // 销毁连接ID
    rdma_destroy_id(listener); // 销毁监听ID
    rdma_destroy_event_channel(ec); // 销毁事件通道
    rdma_freeaddrinfo(res); // 释放地址信息

3.1.8 主函数

运行服务器：

int main() {
    run_server();
    return 0;
}

3.2 客户端代码

保存以下代码为 rdma_client.c：

实现了一个简单的RDMA（Remote Direct Memory Access）客户端，连接到指定的服务器并发送一条消息。RDMA允许计算机通过高效的网络通信方式直接访问彼此的内存，而无需操作系统干预，从而减少延迟并提高数据传输速度。代码使用了RDMA的通信管理API和Infiniband verbs API来实现这个功能。

#include <stdio.h>  // 包含标准输入输出库
#include <stdlib.h>  // 包含标准库
#include <string.h>  // 包含字符串处理库
#include <rdma/rdma_cma.h>  // 包含RDMA连接管理API
#include <infiniband/verbs.h>  // 包含Infiniband verbs API

#define BUFFER_SIZE 1024  // 定义缓冲区大小

// 客户端主函数，接收服务器地址作为参数
void run_client(const char *server) {
    // 声明和初始化变量
    struct rdma_event_channel *ec = rdma_create_event_channel();  // 创建RDMA事件通道
    struct rdma_cm_id *conn = NULL;  // RDMA连接标识符
    struct rdma_addrinfo hints, *res;  // RDMA地址信息结构和指针
    struct ibv_pd *pd = NULL;  // 保护域
    struct ibv_mr *mr = NULL;  // 内存注册区域
    struct ibv_comp_channel *comp_chan = NULL;  // 事件通道
    struct ibv_cq *cq = NULL;  // 完成队列
    struct ibv_qp_init_attr qp_attr;  // 队列对初始化属性
    char buf[BUFFER_SIZE];  // 数据缓冲区
    int ret;  // 返回值变量

    // 初始化hints结构体
    memset(&hints, 0, sizeof(hints));  // 清空结构体
    hints.ai_port_space = RDMA_PS_TCP;  // 设置端口空间为TCP

    // 获取地址信息
    ret = rdma_getaddrinfo(server, "20079", &hints, &res);
    if (ret) {
        perror("rdma_getaddrinfo");  // 打印错误信息
        exit(EXIT_FAILURE);  // 退出程序
    }

    // 创建RDMA标识符
    ret = rdma_create_id(ec, &conn, NULL, RDMA_PS_TCP);
    if (ret) {
        perror("rdma_create_id");
        exit(EXIT_FAILURE);
    }

    // 解析地址
    ret = rdma_resolve_addr(conn, NULL, res->ai_dst_addr, 2000);
    if (ret) {
        perror("rdma_resolve_addr");
        exit(EXIT_FAILURE);
    }

    // 等待地址解析事件
    struct rdma_cm_event *event;
    ret = rdma_get_cm_event(ec, &event);
    if (ret) {
        perror("rdma_get_cm_event");
        exit(EXIT_FAILURE);
    }

    // 检查事件类型
    if (event->event == RDMA_CM_EVENT_ADDR_RESOLVED) {
        rdma_ack_cm_event(event);  // 确认事件
    } else {
        fprintf(stderr, "Unexpected event: %d\n", event->event);
        rdma_ack_cm_event(event);
        exit(EXIT_FAILURE);
    }

    // 解析路由
    ret = rdma_resolve_route(conn, 2000);
    if (ret) {
        perror("rdma_resolve_route");
        exit(EXIT_FAILURE);
    }

    // 等待路由解析事件
    ret = rdma_get_cm_event(ec, &event);
    if (ret) {
        perror("rdma_get_cm_event");
        exit(EXIT_FAILURE);
    }

    // 检查事件类型
    if (event->event == RDMA_CM_EVENT_ROUTE_RESOLVED) {
        rdma_ack_cm_event(event);  // 确认事件
    } else {
        fprintf(stderr, "Unexpected event: %d\n", event->event);
        rdma_ack_cm_event(event);
        exit(EXIT_FAILURE);
    }

    // 分配保护域
    pd = ibv_alloc_pd(conn->verbs);
    if (!pd) {
        perror("ibv_alloc_pd");
        exit(EXIT_FAILURE);
    }

    // 创建完成通道
    comp_chan = ibv_create_comp_channel(conn->verbs);
    if (!comp_chan) {
        perror("ibv_create_comp_channel");
        exit(EXIT_FAILURE);
    }

    // 创建完成队列
    cq = ibv_create_cq(conn->verbs, 10, NULL, comp_chan, 0);
    if (!cq) {
        perror("ibv_create_cq");
        exit(EXIT_FAILURE);
    }

    // 初始化队列对属性
    memset(&qp_attr, 0, sizeof(qp_attr));
    qp_attr.cap.max_send_wr = 1;  // 最大发送请求数
    qp_attr.cap.max_recv_wr = 1;  // 最大接收请求数
    qp_attr.cap.max_send_sge = 1;  // 最大发送SGE数
    qp_attr.cap.max_recv_sge = 1;  // 最大接收SGE数
    qp_attr.send_cq = cq;  // 发送完成队列
    qp_attr.recv_cq = cq;  // 接收完成队列
    qp_attr.qp_type = IBV_QPT_RC;  // 队列对类型

    // 创建队列对
    ret = rdma_create_qp(conn, pd, &qp_attr);
    if (ret) {
        perror("rdma_create_qp");
        exit(EXIT_FAILURE);
    }

    // 注册内存区域
    mr = ibv_reg_mr(pd, buf, BUFFER_SIZE, IBV_ACCESS_LOCAL_WRITE | IBV_ACCESS_REMOTE_WRITE);
    if (!mr) {
        perror("ibv_reg_mr");
        exit(EXIT_FAILURE);
    }

    // 设置连接参数
    struct rdma_conn_param cm_params;
    memset(&cm_params, 0, sizeof(cm_params));
    
    // 发起连接请求
    ret = rdma_connect(conn, &cm_params);
    if (ret) {
        perror("rdma_connect");
        exit(EXIT_FAILURE);
    }

    // 等待连接建立事件
    ret = rdma_get_cm_event(ec, &event);
    if (ret) {
        perror("rdma_get_cm_event");
        exit(EXIT_FAILURE);
    }

    // 检查事件类型
    if (event->event == RDMA_CM_EVENT_ESTABLISHED) {
        rdma_ack_cm_event(event);  // 确认事件
    } else {
        fprintf(stderr, "Unexpected event: %d\n", event->event);
        rdma_ack_cm_event(event);
        exit(EXIT_FAILURE);
    }

    // 准备发送数据
    strcpy(buf, "Hello, RDMA!");
    struct ibv_send_wr wr, *bad_wr = NULL;  // 发送请求和错误请求指针
    struct ibv_sge sge;  // 单次数据描述符
    sge.addr = (uintptr_t)buf;  // 数据缓冲区地址
    sge.length = BUFFER_SIZE;  // 数据长度
    sge.lkey = mr->lkey;  // 本地密钥

    // 初始化发送请求
    wr.wr_id = 0;
    wr.next = NULL;
    wr.sg_list = &sge;  // 数据描述符列表
    wr.num_sge = 1;  // 数据描述符数量
    wr.opcode = IBV_WR_SEND;  // 操作码为发送
    wr.send_flags = IBV_SEND_SIGNALED;  // 发送标志

    // 发送数据
    ret = ibv_post_send(conn->qp, &wr, &bad_wr);
    if (ret) {
        perror("ibv_post_send");
        exit(EXIT_FAILURE);
    }

    // 等待完成队列事件
    struct ibv_wc wc;
    while ((ret = ibv_poll_cq(cq, 1, &wc)) == 0);
    if (ret < 0) {
        perror("ibv_poll_cq");
        exit(EXIT_FAILURE);
    }

    // 检查完成状态
    if (wc.status != IBV_WC_SUCCESS) {
        fprintf(stderr, "Failed status %s (%d) for wr_id %d\n",
                ibv_wc_status_str(wc.status), wc.status, (int) wc.wr_id);
        exit(EXIT_FAILURE);
    }

    // 打印发送消息
    printf("Message sent: %s\n", buf);

    // 断开连接并释放资源
    rdma_disconnect(conn);  // 断开连接
    rdma_destroy_qp(conn);  // 销毁队列对
    ibv_dereg_mr(mr);  // 解除内存注册
    ibv_destroy_cq(cq);  // 销毁完成队列
    ibv_destroy_comp_channel(comp_chan);  // 销毁完成通道
    ibv_dealloc_pd(pd);  // 释放保护域
    rdma_destroy_id(conn);  // 销毁连接标识符
    rdma_destroy_event_channel(ec);  // 销毁事件通道
    rdma_freeaddrinfo(res);  // 释放地址信息
}

// 主函数
int main(int argc, char **argv) {
    if (argc != 2) {
        fprintf(stderr, "Usage: %s <server-address>\n", argv[0]);  // 打印用法信息
        return EXIT_FAILURE;  // 返回失败状态
    }

    run_client(argv[1]);  // 运行客户端
    return 0;  // 返回成功状态
}

3.2.1 初始化和准备工作

1. 创建并初始化RDMA事件通道：

struct rdma_event_channel *ec = rdma_create_event_channel();// 创建RDMA事件通道

2. 定义变量：

    struct rdma_cm_id *conn = NULL;  // RDMA连接标识符
    struct rdma_addrinfo hints, *res;  // RDMA地址信息结构和指针
    struct ibv_pd *pd = NULL;  // 保护域
    struct ibv_mr *mr = NULL;  // 内存注册区域
    struct ibv_comp_channel *comp_chan = NULL;  // 事件通道
    struct ibv_cq *cq = NULL;  // 完成队列
    struct ibv_qp_init_attr qp_attr;  // 队列对初始化属性
    char buf[BUFFER_SIZE];  // 数据缓冲区
    int ret;  // 返回值变量

conn：RDMA连接标识符。
hints和res：用于存储地址信息。
pd、mr、comp_chan和cq：分别代表保护域、内存注册区、完成通道和完成队列。
qp_attr：用于配置QP的初始化属性。
buf：用于存储要发送的数据。
ret：用于存储各个操作的返回值。

3. 配置地址信息：

    // 初始化hints结构体
    memset(&hints, 0, sizeof(hints));  // 清空结构体
    hints.ai_port_space = RDMA_PS_TCP;  // 设置端口空间为TCP

4. 获取RDMA地址：

    // 获取地址信息
    ret = rdma_getaddrinfo(server, "20079", &hints, &res);
    if (ret) {
        perror("rdma_getaddrinfo");  // 打印错误信息
        exit(EXIT_FAILURE);  // 退出程序
    }

3.2.2 创建和解析连接

1. 创建RDMA标识符：

    // 创建RDMA标识符
    ret = rdma_create_id(ec, &conn, NULL, RDMA_PS_TCP);
    if (ret) {
        perror("rdma_create_id");
        exit(EXIT_FAILURE);
    }

2. 解析服务器地址：

    // 解析地址
    ret = rdma_resolve_addr(conn, NULL, res->ai_dst_addr, 2000);
    if (ret) {
        perror("rdma_resolve_addr");
        exit(EXIT_FAILURE);
    }

    // 等待地址解析事件
    struct rdma_cm_event *event;
    ret = rdma_get_cm_event(ec, &event);
    if (ret) {
        perror("rdma_get_cm_event");
        exit(EXIT_FAILURE);
    }

    // 检查事件类型
    if (event->event == RDMA_CM_EVENT_ADDR_RESOLVED) {
        rdma_ack_cm_event(event);  // 确认事件
    } else {
        fprintf(stderr, "Unexpected event: %d\n", event->event);
        rdma_ack_cm_event(event);
        exit(EXIT_FAILURE);
    }

3.2.3 路由解析

    // 解析路由
    ret = rdma_resolve_route(conn, 2000);
    if (ret) {
        perror("rdma_resolve_route");
        exit(EXIT_FAILURE);
    }

    // 等待路由解析事件
    ret = rdma_get_cm_event(ec, &event);
    if (ret) {
        perror("rdma_get_cm_event");
        exit(EXIT_FAILURE);
    }

    // 检查事件类型
    if (event->event == RDMA_CM_EVENT_ROUTE_RESOLVED) {
        rdma_ack_cm_event(event);  // 确认事件
    } else {
        fprintf(stderr, "Unexpected event: %d\n", event->event);
        rdma_ack_cm_event(event);
        exit(EXIT_FAILURE);
    }

3.2.4 资源分配和QP创建

1. 分配保护域、创建完成通道和完成队列：

    // 分配保护域
    pd = ibv_alloc_pd(conn->verbs);
    if (!pd) {
        perror("ibv_alloc_pd");
        exit(EXIT_FAILURE);
    }

    // 创建完成通道
    comp_chan = ibv_create_comp_channel(conn->verbs);
    if (!comp_chan) {
        perror("ibv_create_comp_channel");
        exit(EXIT_FAILURE);
    }

    // 创建完成队列
    cq = ibv_create_cq(conn->verbs, 10, NULL, comp_chan, 0);
    if (!cq) {
        perror("ibv_create_cq");
        exit(EXIT_FAILURE);
    }

2. 初始化并创建QP：

    // 初始化队列对属性
    memset(&qp_attr, 0, sizeof(qp_attr));
    qp_attr.cap.max_send_wr = 1;  // 最大发送请求数
    qp_attr.cap.max_recv_wr = 1;  // 最大接收请求数
    qp_attr.cap.max_send_sge = 1;  // 最大发送SGE数
    qp_attr.cap.max_recv_sge = 1;  // 最大接收SGE数
    qp_attr.send_cq = cq;  // 发送完成队列
    qp_attr.recv_cq = cq;  // 接收完成队列
    qp_attr.qp_type = IBV_QPT_RC;  // 队列对类型

    // 创建队列对
    ret = rdma_create_qp(conn, pd, &qp_attr);
    if (ret) {
        perror("rdma_create_qp");
        exit(EXIT_FAILURE);
    }

3.2.5 内存注册与连接

1. 注册内存区域：

    // 注册内存区域
    mr = ibv_reg_mr(pd, buf, BUFFER_SIZE, IBV_ACCESS_LOCAL_WRITE | IBV_ACCESS_REMOTE_WRITE);
    if (!mr) {
        perror("ibv_reg_mr");
        exit(EXIT_FAILURE);
    }

2. 发起连接请求：

    // 设置连接参数
    struct rdma_conn_param cm_params;
    memset(&cm_params, 0, sizeof(cm_params));
    
    // 发起连接请求
    ret = rdma_connect(conn, &cm_params);
    if (ret) {
        perror("rdma_connect");
        exit(EXIT_FAILURE);
    }

    // 等待连接建立事件
    ret = rdma_get_cm_event(ec, &event);
    if (ret) {
        perror("rdma_get_cm_event");
        exit(EXIT_FAILURE);
    }

    // 检查事件类型
    if (event->event == RDMA_CM_EVENT_ESTABLISHED) {
        rdma_ack_cm_event(event);  // 确认事件
    } else {
        fprintf(stderr, "Unexpected event: %d\n", event->event);
        rdma_ack_cm_event(event);
        exit(EXIT_FAILURE);
    }

3.2.6 数据发送

准备并发送数据：

    // 准备发送数据
    strcpy(buf, "Hello, RDMA!");
    struct ibv_send_wr wr, *bad_wr = NULL;  // 发送请求和错误请求指针
    struct ibv_sge sge;  // 单次数据描述符
    sge.addr = (uintptr_t)buf;  // 数据缓冲区地址
    sge.length = BUFFER_SIZE;  // 数据长度
    sge.lkey = mr->lkey;  // 本地密钥

    // 初始化发送请求
    wr.wr_id = 0;
    wr.next = NULL;
    wr.sg_list = &sge;  // 数据描述符列表
    wr.num_sge = 1;  // 数据描述符数量
    wr.opcode = IBV_WR_SEND;  // 操作码为发送
    wr.send_flags = IBV_SEND_SIGNALED;  // 发送标志

    // 发送数据
    ret = ibv_post_send(conn->qp, &wr, &bad_wr);
    if (ret) {
        perror("ibv_post_send");
        exit(EXIT_FAILURE);
    }

    // 等待完成队列事件
    struct ibv_wc wc;
    while ((ret = ibv_poll_cq(cq, 1, &wc)) == 0);
    if (ret < 0) {
        perror("ibv_poll_cq");
        exit(EXIT_FAILURE);
    }

    // 检查完成状态
    if (wc.status != IBV_WC_SUCCESS) {
        fprintf(stderr, "Failed status %s (%d) for wr_id %d\n",
                ibv_wc_status_str(wc.status), wc.status, (int) wc.wr_id);
        exit(EXIT_FAILURE);
    }

    // 打印发送消息
    printf("Message sent: %s\n", buf);

3.2.7 清理资源

断开连接并释放资源

    // 断开连接并释放资源
    rdma_disconnect(conn);  // 断开连接
    rdma_destroy_qp(conn);  // 销毁队列对
    ibv_dereg_mr(mr);  // 解除内存注册
    ibv_destroy_cq(cq);  // 销毁完成队列
    ibv_destroy_comp_channel(comp_chan);  // 销毁完成通道
    ibv_dealloc_pd(pd);  // 释放保护域
    rdma_destroy_id(conn);  // 销毁连接标识符
    rdma_destroy_event_channel(ec);  // 销毁事件通道
    rdma_freeaddrinfo(res);  // 释放地址信息

3.2.8 主函数：

运行客户端：

// 主函数
int main(int argc, char **argv) {
    if (argc != 2) {
        fprintf(stderr, "Usage: %s <server-address>\n", argv[0]);  // 打印用法信息
        return EXIT_FAILURE;  // 返回失败状态
    }

    run_client(argv[1]);  // 运行客户端
    return 0;  // 返回成功状态
}

3.3 编译和运行

确保先编译服务器和客户端代码：

gcc -o rdma_server rdma_server.c -lrdmacm -libverbs
gcc -o rdma_client rdma_client.c -lrdmacm -libverbs

注意：

如果遇到此错误，因为链接器找不到 ibv_reg_mr_iova2 函数的定义。这个函数可能是你所使用的库版本中不存在的，或者它属于另一个库。

解决方案：

1.确保你安装的库版本是正确的，且包含 ibv_reg_mr_iova2 函数。检查所使用的库版本以及文档，确认该函数确实存在。

2.确保链接器能够找到正确的库。确保你已经正确安装并链接了 rdma-core 包。如果库路径不在默认的搜索路径中，可能需要手动指定库路径。例如：
gcc -o rdma_client rdma_client.c -L/usr/local/lib -lrdmacm -libverbs

3.5 结果

服务器将打印收到的消息：

客户端将打印发送的消息：

可以了解到如何使用 rdma-core 进行基本的 RDMA 连接和数据传输。

四、`rdma-core` 实现更复杂的 RDMA 连接和数据传输操作

为了进一步利用 rdma-core 实现更复杂的 RDMA 连接和数据传输操作，你需要掌握一些基本概念和操作步骤。

4.1 RDMA 操作基础

RDMA 编程涉及多个步骤，包括设备发现、上下文创建、连接建立、数据传输等。

4.2 设备发现与上下文创建

你需要先获取 RDMA 设备列表，并选择一个设备进行上下文创建。

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

// 主函数
int main() {
    struct ibv_device **dev_list;     // 存储设备列表的指针数组
    struct ibv_device *ib_dev;        // 存储选定设备的指针
    struct ibv_context *ctx;          // 存储设备上下文的指针

    // 获取设备列表
    dev_list = ibv_get_device_list(NULL);
    if (!dev_list) {
        // 如果获取设备列表失败，打印错误信息并返回失败状态
        perror("Failed to get devices list");
        return EXIT_FAILURE;
    }

    // 选择第一个设备
    ib_dev = dev_list[0];
    if (!ib_dev) {
        // 如果没有找到任何IB设备，打印错误信息并返回失败状态
        fprintf(stderr, "No IB devices found\n");
        return EXIT_FAILURE;
    }

    // 获取设备的上下文
    ctx = ibv_open_device(ib_dev);
    if (!ctx) {
        // 如果打开设备失败，打印错误信息并返回失败状态
        perror("Failed to open device");
        return EXIT_FAILURE;
    }

    // 打印出成功打开的设备名称
    printf("Device %s opened\n", ib_dev->name);

    // 清理分配的资源
    ibv_close_device(ctx);            // 关闭设备上下文
    ibv_free_device_list(dev_list);   // 释放设备列表

    return EXIT_SUCCESS;              // 返回成功状态
}

展示了如何使用Infiniband Verbs API获取并打开Infiniband设备，适用于需要与Infiniband设备进行低级别交互的应用程序

4.1.1 变量定义

struct ibv_device **dev_list;     // 存储设备列表的指针数组
struct ibv_device *ib_dev;        // 存储选定设备的指针
struct ibv_context *ctx;          // 存储设备上下文的指针

声明了三个指针变量：

dev_list：用于存储设备列表。
ib_dev：用于存储选定的设备。
ctx：用于存储设备的上下文信息。

4.1.2 获取设备列表：

dev_list = ibv_get_device_list(NULL);
if (!dev_list) {
     perror("Failed to get devices list");
     return EXIT_FAILURE;
}

调用ibv_get_device_list函数获取设备列表，如果获取失败，打印错误信息并返回失败状态。

4.1.3 选择设备：

ib_dev = dev_list[0];
if (!ib_dev) {
    fprintf(stderr, "No IB devices found\n");
    return EXIT_FAILURE;
}

选择设备列表中的第一个设备。如果设备列表为空，打印错误信息并返回失败状态。

4.1.4 获取设备上下文

ctx = ibv_open_device(ib_dev);
if (!ctx) {
    perror("Failed to open device");
    return EXIT_FAILURE;
}

调用ibv_open_device函数获取选定设备的上下文。如果获取失败，打印错误信息并返回失败状态。

4.1.5 打印设备名称并且清理资源

printf("Device %s opened\n", ib_dev->name);
ibv_close_device(ctx);            // 关闭设备上下文
ibv_free_device_list(dev_list);   // 释放设备列表

打印出成功打开的设备名称，关闭设备上下文并释放设备列表，清理资源。

4.1.6 编译运行：

将提供的代码保存到一个文件中，例如ibv_example.c。

使用GCC编译器来编译代码，并链接Infiniband库：

gcc -o ibv_example ibv_example.c -libverbs

运行编译生成的可执行文件：

./ibv_example

4.1.7 示例输出：

假设系统中有可用的Infiniband设备，并且代码运行成功，输出可能如下：

Device rxe0 opened

如果系统中没有Infiniband设备，输出可能如下：

No IB devices found

如果在获取设备列表或打开设备时发生错误，输出可能如下：

Failed to get devices list: <error_description>
Failed to open device: <error_description>

此时我们输出则显示有可用的Infiniband设备：

4.3 内存注册与队列对创建

为了传输数据，你需要注册内存区域并创建发送和接收队列。

#include <stdio.h>                       // 包含标准输入输出头文件
#include <stdlib.h>                      // 包含标准库头文件
#include <string.h>                      // 包含字符串操作头文件
#include <infiniband/verbs.h>            // 包含InfiniBand verbs库头文件

#define BUFFER_SIZE 1024                 // 定义缓冲区大小为1024字节

int main() {
    struct ibv_device **dev_list;        // 定义设备列表指针，指向InfiniBand设备数组
    struct ibv_device *ib_dev;           // 定义设备指针，指向单个InfiniBand设备
    struct ibv_context *ctx;             // 定义上下文指针，表示设备上下文
    struct ibv_pd *pd;                   // 定义保护域指针
    struct ibv_mr *mr;                   // 定义内存注册区域指针
    struct ibv_cq *cq;                   // 定义完成队列指针
    struct ibv_qp *qp;                   // 定义队列对指针
    struct ibv_qp_init_attr qp_attr;     // 定义队列对初始化属性结构
    char buf[BUFFER_SIZE];               // 定义缓冲区

    // 获取设备列表并打开设备
    dev_list = ibv_get_device_list(NULL);                  // 获取所有InfiniBand设备列表
    ib_dev = dev_list[0];                                  // 选择第一个设备
    ctx = ibv_open_device(ib_dev);                         // 打开设备并获取设备上下文

    // 创建保护域
    pd = ibv_alloc_pd(ctx);                                // 为设备上下文分配保护域
    if (!pd) {                                             // 检查保护域是否分配成功
        perror("ibv_alloc_pd");                            // 如果失败，打印错误信息
        return EXIT_FAILURE;                               // 返回失败状态
    }

    // 注册内存区域
    mr = ibv_reg_mr(pd, buf, BUFFER_SIZE, IBV_ACCESS_LOCAL_WRITE | IBV_ACCESS_REMOTE_WRITE);  // 注册内存区域
    if (!mr) {                                             // 检查内存区域是否注册成功
        perror("ibv_reg_mr");                              // 如果失败，打印错误信息
        return EXIT_FAILURE;                               // 返回失败状态
    }

    // 创建完成队列
    cq = ibv_create_cq(ctx, 10, NULL, NULL, 0);            // 创建一个完成队列，最多包含10个完成元素
    if (!cq) {                                             // 检查完成队列是否创建成功
        perror("ibv_create_cq");                           // 如果失败，打印错误信息
        return EXIT_FAILURE;                               // 返回失败状态
    }

    // 设置队列对属性
    memset(&qp_attr, 0, sizeof(qp_attr));                  // 将队列对初始化属性结构清零
    qp_attr.cap.max_send_wr = 1;                           // 设置最大发送请求数为1
    qp_attr.cap.max_recv_wr = 1;                           // 设置最大接收请求数为1
    qp_attr.cap.max_send_sge = 1;                          // 设置单个发送请求最大散播聚合元素数为1
    qp_attr.cap.max_recv_sge = 1;                          // 设置单个接收请求最大散播聚合元素数为1
    qp_attr.send_cq = cq;                                  // 关联发送完成队列
    qp_attr.recv_cq = cq;                                  // 关联接收完成队列
    qp_attr.qp_type = IBV_QPT_RC;                          // 设置队列对类型为可靠连接

    // 创建队列对
    qp = ibv_create_qp(pd, &qp_attr);                      // 创建队列对
    if (!qp) {                                             // 检查队列对是否创建成功
        perror("ibv_create_qp");                           // 如果失败，打印错误信息
        return EXIT_FAILURE;                               // 返回失败状态
    }

    printf("Queue Pair created successfully\n");           // 打印成功创建队列对的消息

    // 清理
    ibv_destroy_qp(qp);                                    // 销毁队列对
    ibv_destroy_cq(cq);                                    // 销毁完成队列
    ibv_dereg_mr(mr);                                      // 取消注册内存区域
    ibv_dealloc_pd(pd);                                    // 释放保护域
    ibv_close_device(ctx);                                 // 关闭设备
    ibv_free_device_list(dev_list);                        // 释放设备列表

    return EXIT_SUCCESS;                                   // 返回成功状态
}

演示如何使用libibverbs库创建并初始化一个InfiniBand队列对（Queue Pair，QP）。

4.3.1 获取InfiniBand设备列表并打开设备:（此过程就是4.2介绍的过程）

dev_list = ibv_get_device_list(NULL);
ib_dev = dev_list[0];
ctx = ibv_open_device(ib_dev);

4.3.2 创建保护域（Protection Domain，PD）:

pd = ibv_alloc_pd(ctx);

4.3.3 注册内存区域:

mr = ibv_reg_mr(pd, buf, BUFFER_SIZE, IBV_ACCESS_LOCAL_WRITE | IBV_ACCESS_REMOTE_WRITE);

4.3.4 创建完成队列（Completion Queue，CQ）:

cq = ibv_create_cq(ctx, 10, NULL, NULL, 0);

4.3.5 设置队列对属性并创建队列对:

memset(&qp_attr, 0, sizeof(qp_attr));
qp_attr.cap.max_send_wr = 1;
qp_attr.cap.max_recv_wr = 1;
qp_attr.cap.max_send_sge = 1;
qp_attr.cap.max_recv_sge = 1;
qp_attr.send_cq = cq;
qp_attr.recv_cq = cq;
qp_attr.qp_type = IBV_QPT_RC;

qp = ibv_create_qp(pd, &qp_attr);

4.3.6 清理资源:

程序结束前清理分配的资源。

ibv_destroy_qp(qp);
ibv_destroy_cq(cq);
ibv_dereg_mr(mr);
ibv_dealloc_pd(pd);
ibv_close_device(ctx);
ibv_free_device_list(dev_list);

4.3.7 演示效果

这表示队列对（Queue Pair）成功创建。程序会在执行后清理所有资源，确保没有资源泄漏。

4.4 建立 RDMA 连接

你需要使用 rdma_cm 库来建立 RDMA 连接。这包括服务器和客户端之间的连接建立。

1. 服务器端代码（与3.1代码略有不同）

与3.1最大差异体现在资源释放更加全面，确保在出错时也能释放所有已分配的资源。

#include <stdio.h>       // 包含标准输入输出库
#include <stdlib.h>      // 包含标准库
#include <string.h>      // 包含字符串处理库
#include <rdma/rdma_cma.h>  // 包含RDMA连接管理API

#define BUFFER_SIZE 1024  // 定义缓冲区大小

void run_server() {
    // 创建RDMA事件通道，用于接收RDMA事件
    struct rdma_event_channel *ec = rdma_create_event_channel();
    struct rdma_cm_id *listener = NULL, *conn = NULL;  // RDMA连接标识符
    struct rdma_addrinfo hints, *res;  // RDMA地址信息结构和指针
    struct ibv_pd *pd = NULL;  // 保护域
    struct ibv_mr *mr = NULL;  // 内存注册区域
    struct ibv_comp_channel *comp_chan = NULL;  // 完成事件通道
    struct ibv_cq *cq = NULL;  // 完成队列
    struct ibv_qp_init_attr qp_attr;  // 队列对初始化属性
    char buf[BUFFER_SIZE];  // 数据缓冲区
    int ret;  // 返回值变量

    // 配置地址信息
    memset(&hints, 0, sizeof(hints));  // 清空结构体
    hints.ai_flags = RAI_PASSIVE;  // 设置为被动模式
    hints.ai_port_space = RDMA_PS_TCP;  // 设置端口空间为TCP

    // 获取地址信息
    ret = rdma_getaddrinfo(NULL, "20079", &hints, &res);
    if (ret) {
        perror("rdma_getaddrinfo");  // 打印错误信息
        exit(EXIT_FAILURE);  // 退出程序
    }

    // 创建RDMA标识符
    ret = rdma_create_id(ec, &listener, NULL, RDMA_PS_TCP);
    if (ret) {
        perror("rdma_create_id");
        rdma_freeaddrinfo(res);  // 释放地址信息
        exit(EXIT_FAILURE);
    }

    // 绑定地址
    ret = rdma_bind_addr(listener, res->ai_src_addr);
    if (ret) {
        perror("rdma_bind_addr");
        rdma_destroy_id(listener);  // 销毁RDMA标识符
        rdma_freeaddrinfo(res);  // 释放地址信息
        exit(EXIT_FAILURE);
    }

    // 监听连接请求
    ret = rdma_listen(listener, 0);
    if (ret) {
        perror("rdma_listen");
        rdma_destroy_id(listener);  // 销毁RDMA标识符
        rdma_freeaddrinfo(res);  // 释放地址信息
        exit(EXIT_FAILURE);
    }

    printf("Server is listening on port 20079...\n");

    // 等待连接请求事件
    struct rdma_cm_event *event;
    ret = rdma_get_cm_event(ec, &event);
    if (ret) {
        perror("rdma_get_cm_event");
        rdma_destroy_id(listener);  // 销毁RDMA标识符
        rdma_freeaddrinfo(res);  // 释放地址信息
        exit(EXIT_FAILURE);
    }

    // 处理连接请求事件
    if (event->event == RDMA_CM_EVENT_CONNECT_REQUEST) {
        conn = event->id;  // 获取连接标识符
        rdma_ack_cm_event(event);  // 确认事件
    } else {
        fprintf(stderr, "Unexpected event: %d\n", event->event);
        rdma_ack_cm_event(event);  // 确认事件
        rdma_destroy_id(listener);  // 销毁RDMA标识符
        rdma_freeaddrinfo(res);  // 释放地址信息
        exit(EXIT_FAILURE);
    }

    // 分配保护域
    pd = ibv_alloc_pd(conn->verbs);
    if (!pd) {
        perror("ibv_alloc_pd");
        rdma_destroy_id(conn);  // 销毁连接标识符
        rdma_destroy_id(listener);  // 销毁监听标识符
        rdma_freeaddrinfo(res);  // 释放地址信息
        exit(EXIT_FAILURE);
    }

    // 创建完成事件通道
    comp_chan = ibv_create_comp_channel(conn->verbs);
    if (!comp_chan) {
        perror("ibv_create_comp_channel");
        ibv_dealloc_pd(pd);  // 释放保护域
        rdma_destroy_id(conn);  // 销毁连接标识符
        rdma_destroy_id(listener);  // 销毁监听标识符
        rdma_freeaddrinfo(res);  // 释放地址信息
        exit(EXIT_FAILURE);
    }

    // 创建完成队列
    cq = ibv_create_cq(conn->verbs, 10, NULL, comp_chan, 0);
    if (!cq) {
        perror("ibv_create_cq");
        ibv_destroy_comp_channel(comp_chan);  // 销毁完成事件通道
        ibv_dealloc_pd(pd);  // 释放保护域
        rdma_destroy_id(conn);  // 销毁连接标识符
        rdma_destroy_id(listener);  // 销毁监听标识符
        rdma_freeaddrinfo(res);  // 释放地址信息
        exit(EXIT_FAILURE);
    }

    // 配置队列对属性
    memset(&qp_attr, 0, sizeof(qp_attr));
    qp_attr.cap.max_send_wr = 1;  // 最大发送请求数
    qp_attr.cap.max_recv_wr = 1;  // 最大接收请求数
    qp_attr.cap.max_send_sge = 1;  // 最大发送SGE数
    qp_attr.cap.max_recv_sge = 1;  // 最大接收SGE数
    qp_attr.send_cq = cq;  // 发送完成队列
    qp_attr.recv_cq = cq;  // 接收完成队列
    qp_attr.qp_type = IBV_QPT_RC;  // 队列对类型

    // 创建队列对
    ret = rdma_create_qp(conn, pd, &qp_attr);
    if (ret) {
        perror("rdma_create_qp");
        ibv_destroy_cq(cq);  // 销毁完成队列
        ibv_destroy_comp_channel(comp_chan);  // 销毁完成事件通道
        ibv_dealloc_pd(pd);  // 释放保护域
        rdma_destroy_id(conn);  // 销毁连接标识符
        rdma_destroy_id(listener);  // 销毁监听标识符
        rdma_freeaddrinfo(res);  // 释放地址信息
        exit(EXIT_FAILURE);
    }

    // 注册内存区域
    mr = ibv_reg_mr(pd, buf, BUFFER_SIZE, IBV_ACCESS_LOCAL_WRITE | IBV_ACCESS_REMOTE_WRITE);
    if (!mr) {
        perror("ibv_reg_mr");
        rdma_destroy_qp(conn);  // 销毁队列对
        ibv_destroy_cq(cq);  // 销毁完成队列
        ibv_destroy_comp_channel(comp_chan);  // 销毁完成事件通道
        ibv_dealloc_pd(pd);  // 释放保护域
        rdma_destroy_id(conn);  // 销毁连接标识符
        rdma_destroy_id(listener);  // 销毁监听标识符
        rdma_freeaddrinfo(res);  // 释放地址信息
        exit(EXIT_FAILURE);
    }

    // 设置接收请求
    struct ibv_recv_wr wr, *bad_wr = NULL;
    struct ibv_sge sge;
    sge.addr = (uintptr_t)buf;
    sge.length = BUFFER_SIZE;
    sge.lkey = mr->lkey;

    wr.wr_id = 0;
    wr.next = NULL;
    wr.sg_list = &sge;
    wr.num_sge = 1;

    // 发起接收请求
    ret = ibv_post_recv(conn->qp, &wr, &bad_wr);
    if (ret) {
        perror("ibv_post_recv");
        ibv_dereg_mr(mr);  // 解除内存注册
        rdma_destroy_qp(conn);  // 销毁队列对
        ibv_destroy_cq(cq);  // 销毁完成队列
        ibv_destroy_comp_channel(comp_chan);  // 销毁完成事件通道
        ibv_dealloc_pd(pd);  // 释放保护域
        rdma_destroy_id(conn);  // 销毁连接标识符
        rdma_destroy_id(listener);  // 销毁监听标识符
        rdma_freeaddrinfo(res);  // 释放地址信息
        exit(EXIT_FAILURE);
    }

    // 接受连接请求
    struct rdma_conn_param cm_params;
    memset(&cm_params, 0, sizeof(cm_params));
    ret = rdma_accept(conn, &cm_params);
    if (ret) {
        perror("rdma_accept");
        rdma_disconnect(conn);  // 断开连接
        ibv_dereg_mr(mr);  // 解除内存注册
        rdma_destroy_qp(conn);  // 销毁队列对
        ibv_destroy_cq(cq);  // 销毁完成队列
        ibv_destroy_comp_channel(comp_chan);  // 销毁完成事件通道
        ibv_dealloc_pd(pd);  // 释放保护域
        rdma_destroy_id(conn);  // 销毁连接标识符
        rdma_destroy_id(listener);  // 销毁监听标识符
        rdma_freeaddrinfo(res);  // 释放地址信息
        exit(EXIT_FAILURE);
    }

    // 轮询完成队列，等待数据接收
    struct ibv_wc wc;
    while ((ret = ibv_poll_cq(cq, 1, &wc)) == 0);
    if (ret < 0) {
        perror("ibv_poll_cq");
        rdma_disconnect(conn);  // 断开连接
        ibv_dereg_mr(mr);  // 解除内存注册
        rdma_destroy_qp(conn);  // 销毁队列对
        ibv_destroy_cq(cq);  // 销毁完成队列
        ibv_destroy_comp_channel(comp_chan);  // 销毁完成事件通道
        ibv_dealloc_pd(pd);  // 释放保护域
        rdma_destroy_id(conn);  // 销毁连接标识符
        rdma_destroy_id(listener);  // 销毁监听标识符
        rdma_freeaddrinfo(res);  // 释放地址信息
        exit(EXIT_FAILURE);
    }

    // 检查完成状态
    if (wc.status != IBV_WC_SUCCESS) {
        fprintf(stderr, "Failed status %s (%d) for wr_id %d\n",
                ibv_wc_status_str(wc.status), wc.status, (int) wc.wr_id);
        rdma_disconnect(conn);  // 断开连接
        ibv_dereg_mr(mr);  // 解除内存注册
        rdma_destroy_qp(conn);  // 销毁队列对
        ibv_destroy_cq(cq);  // 销毁完成队列
        ibv_destroy_comp_channel(comp_chan);  // 销毁完成事件通道
        ibv_dealloc_pd(pd);  // 释放保护域
        rdma_destroy_id(conn);  // 销毁连接标识符
        rdma_destroy_id(listener);  // 销毁监听标识符
        rdma_freeaddrinfo(res);  // 释放地址信息
        exit(EXIT_FAILURE);
    }

    // 打印接收到的消息
    printf("Received message: %s\n", buf);

    // 清理资源
    rdma_disconnect(conn);  // 断开连接
    rdma_destroy_qp(conn);  // 销毁队列对
    ibv_dereg_mr(mr);  // 解除内存注册
    ibv_destroy_cq(cq);  // 销毁完成队列
    ibv_destroy_comp_channel(comp_chan);  // 销毁完成事件通道
    ibv_dealloc_pd(pd);  // 释放保护域
    rdma_destroy_id(conn);  // 销毁连接标识符
    rdma_destroy_id(listener);  // 销毁监听标识符
    rdma_destroy_event_channel(ec);  // 销毁事件通道
    rdma_freeaddrinfo(res);  // 释放地址信息
}

int main() {
    run_server();  // 运行服务器
    return 0;
}

2. 客户端代码（与3.2代码相同）

保存以下代码为 rdma_client.c：

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <rdma/rdma_cma.h>
#include <infiniband/verbs.h>

#define BUFFER_SIZE 1024

void run_client(const char *server) {
    struct rdma_event_channel *ec = rdma_create_event_channel();
    struct rdma_cm_id *conn = NULL;
    struct rdma_addrinfo hints, *res;
    struct ibv_pd *pd = NULL;
    struct ibv_mr *mr = NULL;
    struct ibv_comp_channel *comp_chan = NULL;
    struct ibv_cq *cq = NULL;
    struct ibv_qp_init_attr qp_attr;
    char buf[BUFFER_SIZE];
    int ret;

    // 配置地址信息
    memset(&hints, 0, sizeof(hints));
    hints.ai_port_space = RDMA_PS_TCP;

    // 获取地址信息
    ret = rdma_getaddrinfo(server, "20079", &hints, &res);
    if (ret) {
        perror("rdma_getaddrinfo");
        exit(EXIT_FAILURE);
    }

    // 创建 RDMA 连接标识符
    ret = rdma_create_id(ec, &conn, NULL, RDMA_PS_TCP);
    if (ret) {
        perror("rdma_create_id");
        exit(EXIT_FAILURE);
    }

    // 解析地址
    ret = rdma_resolve_addr(conn, NULL, res->ai_dst_addr, 2000);
    if (ret) {
        perror("rdma_resolve_addr");
        exit(EXIT_FAILURE);
    }

    struct rdma_cm_event *event;
    ret = rdma_get_cm_event(ec, &event);
    if (ret) {
        perror("rdma_get_cm_event");
        exit(EXIT_FAILURE);
    }

    if (event->event == RDMA_CM_EVENT_ADDR_RESOLVED) {
        rdma_ack_cm_event(event);
    } else {
        fprintf(stderr, "Unexpected event: %d\n", event->event);
        rdma_ack_cm_event(event);
        exit(EXIT_FAILURE);
    }

    // 解析路由
    ret = rdma_resolve_route(conn, 2000);
    if (ret) {
        perror("rdma_resolve_route");
        exit(EXIT_FAILURE);
    }

    ret = rdma_get_cm_event(ec, &event);
    if (ret) {
        perror("rdma_get_cm_event");
        exit(EXIT_FAILURE);
    }

    if (event->event == RDMA_CM_EVENT_ROUTE_RESOLVED) {
        rdma_ack_cm_event(event);
    } else {
        fprintf(stderr, "Unexpected event: %d\n", event->event);
        rdma_ack_cm_event(event);
        exit(EXIT_FAILURE);
    }

    // 分配保护域
    pd = ibv_alloc_pd(conn->verbs);
    if (!pd) {
        perror("ibv_alloc_pd");
        exit(EXIT_FAILURE);
    }

    // 创建完成事件通道
    comp_chan = ibv_create_comp_channel(conn->verbs);
    if (!comp_chan) {
        perror("ibv_create_comp_channel");
        exit(EXIT_FAILURE);
    }

    // 创建完成队列
    cq = ibv_create_cq(conn->verbs, 10, NULL, comp_chan, 0);
    if (!cq) {
        perror("ibv_create_cq");
        exit(EXIT_FAILURE);
    }

    // 配置队列对属性
    memset(&qp_attr, 0, sizeof(qp_attr));
    qp_attr.cap.max_send_wr = 1;
    qp_attr.cap.max_recv_wr = 1;
    qp_attr.cap.max_send_sge = 1;
    qp_attr.cap.max_recv_sge = 1;
    qp_attr.send_cq = cq;
    qp_attr.recv_cq = cq;
    qp_attr.qp_type = IBV_QPT_RC;

    // 创建队列对
    ret = rdma_create_qp(conn, pd, &qp_attr);
    if (ret) {
        perror("rdma_create_qp");
        exit(EXIT_FAILURE);
    }

    // 注册内存区域
    mr = ibv_reg_mr(pd, buf, BUFFER_SIZE, IBV_ACCESS_LOCAL_WRITE | IBV_ACCESS_REMOTE_WRITE);
    if (!mr) {
        perror("ibv_reg_mr");
        exit(EXIT_FAILURE);
    }

    // 连接到服务器
    struct rdma_conn_param cm_params;
    memset(&cm_params, 0, sizeof(cm_params));
    ret = rdma_connect(conn, &cm_params);
    if (ret) {
        perror("rdma_connect");
        exit(EXIT_FAILURE);
    }

    // 等待连接建立事件
    ret = rdma_get_cm_event(ec, &event);
    if (ret) {
        perror("rdma_get_cm_event");
        exit(EXIT_FAILURE);
    }

    if (event->event == RDMA_CM_EVENT_ESTABLISHED) {
        rdma_ack_cm_event(event);
    } else {
        fprintf(stderr, "Unexpected event: %d\n", event->event);
        rdma_ack_cm_event(event);
        exit(EXIT_FAILURE);
    }

    // 准备发送数据
    strcpy(buf, "Hello, RDMA!");
    struct ibv_send_wr wr, *bad_wr = NULL;
    struct ibv_sge sge;
    sge.addr = (uintptr_t)buf;
    sge.length = BUFFER_SIZE;
    sge.lkey = mr->lkey;

    wr.wr_id = 0;
    wr.next = NULL;
    wr.sg_list = &sge;
    wr.num_sge = 1;
    wr.opcode = IBV_WR_SEND;
    wr.send_flags = IBV_SEND_SIGNALED;

    // 发送数据
    ret = ibv_post_send(conn->qp, &wr, &bad_wr);
    if (ret) {
        perror("ibv_post_send");
        exit(EXIT_FAILURE);
    }

    // 轮询完成队列
    struct ibv_wc wc;
    while ((ret = ibv_poll_cq(cq, 1, &wc)) == 0);
    if (ret < 0) {
        perror("ibv_poll_cq");
        exit(EXIT_FAILURE);
    }

    if (wc.status != IBV_WC_SUCCESS) {
        fprintf(stderr, "Failed status %s (%d) for wr_id %d\n",
                ibv_wc_status_str(wc.status), wc.status, (int) wc.wr_id);
        exit(EXIT_FAILURE);
    }

    printf("Message sent: %s\n", buf);

    // 清理资源
    rdma_disconnect(conn);
    rdma_destroy_qp(conn);
    ibv_dereg_mr(mr);
    ibv_destroy_cq(cq);
    ibv_destroy_comp_channel(comp_chan);
    ibv_dealloc_pd(pd);
    rdma_destroy_id(conn);
    rdma_destroy_event_channel(ec);
    rdma_freeaddrinfo(res);
}

int main(int argc, char **argv) {
    if (argc != 2) {
        fprintf(stderr, "Usage: %s <server-address>\n", argv[0]);
        return EXIT_FAILURE;
    }

    run_client(argv[1]);
    return 0;
}

4.5 编译和运行

确保先编译服务器和客户端代码：

gcc -o rdma_server rdma_server.c -lrdmacm -libverbs
gcc -o rdma_client rdma_client.c -lrdmacm -libverbs

注意：

如果遇到此错误，因为链接器找不到 ibv_reg_mr_iova2 函数的定义。这个函数可能是你所使用的库版本中不存在的，或者它属于另一个库。

解决方案：

1.确保你安装的库版本是正确的，且包含 ibv_reg_mr_iova2 函数。检查所使用的库版本以及文档，确认该函数确实存在。

2.确保链接器能够找到正确的库。确保你已经正确安装并链接了 rdma-core 包。如果库路径不在默认的搜索路径中，可能需要手动指定库路径。例如：
gcc -o rdma_client rdma_client.c -L/usr/local/lib -lrdmacm -libverbs

首先启动服务器：

./rdma_server

然后在另一个终端中启动客户端，替换 <server-address> 为服务器的实际 IP 地址：

./rdma_client <server-address>

4.6 结果

服务器将打印收到的消息：

客户端将打印发送的消息：

可以了解到如何使用 rdma-core 进行基本的 RDMA 连接和数据传输。

五、验证RDMA 在高性能和低延迟场景中的应用价值

验证 RDMA（远程直接内存访问）在高性能和低延迟场景中的应用价值，可以通过一系列实验和测试来进行。下面是一些常见的方法和指标，用于验证和评估 RDMA 系统的性能和延迟。

5.1 性能测试方法

5.1.1 吞吐量测试：

目的：评估系统的数据传输速度。

方法：在一段时间内持续传输数据，统计传输的数据量（如每秒传输的字节数）。

工具：可以使用 Iperf 等网络性能测试工具，或编写自定义测试脚本。

5.1.2 延迟测试：

目的：评估系统的数据传输延迟。

方法：测量发送方发送数据包到接收方接收到数据包之间的时间。

工具：可以使用 ping、qperf 等工具，或通过编写自定义代码测量 RTT（往返时间）。

5.2.3 零拷贝效应：

目的：评估零拷贝技术对系统性能的提升。

方法：比较启用和禁用零拷贝技术情况下的吞吐量和延迟。

工具：自定义测试代码和性能分析工具。

5.2.4 资源使用率分析：

目的：评估系统的资源消耗情况，包括 CPU 使用率、内存占用等。

方法：监控系统在数据传输过程中的资源使用情况。

工具：top、htop、vmstat 等系统监控工具。

5.2 实验设计

5.2.1 测试环境搭建：

搭建一个包含服务器和客户端的测试环境，可以使用物理机或虚拟机。

配置网络，使其支持 RDMA 通信。

5.2.2 基准测试：

选择传统的 TCP/IP 网络通信作为对比基准。

运行相同的测试用例，对比 RDMA 和 TCP/IP 在吞吐量和延迟上的性能差异。

5.2.3 压力测试：

通过增加并发连接数和数据包大小，测试 RDMA 系统在高负载下的表现。

观察系统在高负载下的稳定性和性能变化。

5.3 测试指标

5.3.1 吞吐量（Throughput）：

单位：Gbps 或 MB/s。

高吞吐量表明系统能够在单位时间内传输大量数据。

5.3.2 延迟（Latency）：

单位：微秒（µs）或毫秒（ms）。

低延迟表明系统能够快速传输数据，适用于实时应用。

5.3.3 资源使用率（Resource Utilization）：

包括 CPU 使用率、内存占用、网络带宽利用率等。

低资源使用率表明系统在高效运行。

5.4 实验结果分析

通过上述方法和指标，可以得到如下结论：

1.高吞吐量：如果 RDMA 系统在吞吐量测试中表现优异，传输速率显著高于传统 TCP/IP 网络，则表明 RDMA 系统能够在单位时间内传输更多的数据。

2. 低延迟：如果 RDMA 系统在延迟测试中表现出显著低于传统网络的延迟，则表明 RDMA 系统能够较快地传输数据，适用于对时间敏感的应用场景。

3. 低资源消耗：如果 RDMA 系统在资源使用率分析中表现出较低的 CPU 和内存消耗，则表明 RDMA 系统在处理数据传输时较为高效。

5.5 实验示例

环境：两台配置相同的服务器，连接在同一 RDMA 支持的网络中。
基准测试：
- 使用 TCP/IP 进行数据传输，记录吞吐量和延迟。
- 使用 RDMA 进行相同的数据传输，记录吞吐量和延迟。
结果：
- TCP/IP 吞吐量：5 Gbps；延迟：500 微秒。
- RDMA 吞吐量：20 Gbps；延迟：50 微秒。
资源使用率：
- TCP/IP CPU 使用率：80%；内存占用：1 GB。
- RDMA CPU 使用率：20%；内存占用：500 MB。

通过对比，可以得出结论：RDMA 系统在吞吐量上有显著提升，延迟大幅降低，同时资源消耗也更低。这些结果验证了 RDMA 在高性能和低延迟场景中的应用价值。