RDMA高性能网络通信实践

一、背景介绍

远程直接内存访问（RDMA）技术通过绕过操作系统内核和CPU直接访问远程内存，实现了超低延迟、高吞吐量的网络通信。该技术广泛应用于高性能计算、分布式存储和机器学习等领域。本文通过一个完整的代码示例，演示如何利用RDMA核心组件（QP、MR、CQ等）实现跨节点内存直接读写。

二、方法设计

A.实现方案

1. 控制平面：使用TCP协议交换RDMA连接参数
2. 数据平面：基于IB Verbs接口实现零拷贝传输
3. 混合模式：客户端主动写入，服务端被动读取

B.关键技术点

• 内存注册机制实现安全访问
• QP状态机转换确保通信可靠性
• 完成队列轮询实现异步通知
• 端到端流控通过TCP协议实现

三、代码及注释

/*----------------------------- 头文件包含 -----------------------------*/
// 标准库和网络相关头文件
#include <netdb.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/socket.h>
#include <unistd.h>
#include <assert.h>
#include <errno.h>
#include <sys/time.h>
#include <byteswap.h>// RDMA相关头文件
#include <arpa/inet.h>
#include <infiniband/verbs.h>#ifdef USE_VACC
#include "vaccrt.h"
#include <vaccrt_mem_management.h>
#endif#ifdef USE_CUDA
#include <cuda.h>
#include <cuda_runtime.h>
#endif/**********************************************************************A.关键概念解释：1.保护域（PD）：资源隔离单元，所有资源（QP、MR等）必须属于某个PD2.内存区域（MR）：注册的内存区域，只有注册的内存才能用于RDMA操作3.队列对（QP）：包含发送队列和接收队列，是通信的基本单元4.工作请求（WR）：描述要执行的操作（发送/接收/RDMA读写）5.完成队列（CQ）：用于通知操作完成状态6.QP状态转换：a.INIT：初始状态，设置基本参数b.RTR（Ready to Receive）：准备好接收数据c.RTS（Ready to Send）：准备好发送数据B.程序流程总结：1.通过TCP交换RDMA连接参数2.初始化IB资源（PD、CQ、MR、QP）3.交换QP信息（地址、密钥等）4.进行QP状态转换（INIT->RTR->RTS）5.执行RDMA写/读操作6.轮询完成队列确认操作完成7.清理资源C.使用说明:1.编译命令: gcc -o cuda -DUSE_CUDA -ggdb main.c -pthread -libverbs \-I /usr/local/cuda/include \-L /usr/local/cuda/lib64 -Wl,-rpath=/usr/local/cuda/lib64 \-lcudart -lcudadevrt -lcuda  2.服务端  : ./cuda 192.168.1.100 mlx5_03.客户端  : ./cuda 192.168.1.101 mlx5_0 192.168.1.100 ***********************************************************************//*----------------------------- 全局配置 -----------------------------*/
#define MAX_POLL_CQ_TIMEOUT 6000    // CQ轮询超时时间（毫秒）
#define MSG "Hello,World"           // 要传输的测试消息
#define MSG_SIZE (64<<10)// 配置参数结构体
struct config {const char *dev;      // IB设备名称char *local_addr;     // 本地IP地址u_int32_t port;       // TCP端口号int ib_port;          // IB端口号（默认1）int gid_idx;          // GID索引（-1表示不使用RoCEv2）
} config = { NULL, NULL, 12025, 1, -1 };/*----------------------------- 资源结构体 -----------------------------*/
// 包含所有RDMA相关资源
struct resources {struct ibv_context *ctx;        // IB上下文struct ibv_pd *pd;              // 保护域（Protection Domain）struct ibv_cq *cq;              // 完成队列（Completion Queue）struct ibv_qp *qp;              // 队列对（Queue Pair）struct ibv_mr *mr;              // 内存区域（Memory Region）void *buf;                      // 数据缓冲区指针int sock;                       // TCP套接字uint64_t remote_addr;           // 远程内存地址uint32_t remote_rkey;           // 远程内存访问密钥struct ibv_port_attr port_attr; // IB端口属性
};/*----------------------------- 辅助函数 -----------------------------*/
/*** 建立TCP连接（客户端）或监听（服务端）* @param server 本地地址（服务端模式时使用）* @param port TCP端口号* @param remote_addr 远程地址（客户端模式时使用）* @return 成功返回套接字fd，失败返回-1*/
int sock_connect(const char *server, int port,const char *remote_addr) {struct addrinfo hints = { .ai_family = AF_INET, .ai_socktype = SOCK_STREAM };struct addrinfo *res, *p;int sock = -1;char port_str[6];const char * p_addr=server;if(remote_addr) p_addr=remote_addr;sprintf(port_str, "%d", port);if (getaddrinfo(p_addr, port_str, &hints, &res)) return -1;for (p = res; p; p = p->ai_next) {sock = socket(p->ai_family, p->ai_socktype, p->ai_protocol);int reuse = 1;if(setsockopt(sock, SOL_SOCKET, SO_REUSEADDR, &reuse, sizeof(reuse))) return -1;        if (sock < 0) continue;if (remote_addr) {if (connect(sock, p->ai_addr, p->ai_addrlen)) { close(sock); sock = -1; }else break;} else {if (bind(sock, p->ai_addr, p->ai_addrlen) || listen(sock, 1)) { close(sock); sock = -1; }else { sock = accept(sock, NULL, 0); break; }}}freeaddrinfo(res);return sock;
}/*** 通过TCP同步交换数据* @return 成功返回0，失败返回-1*/
int sock_sync(int sock, int size, void *local, void *remote) {if (write(sock, local, size) != size) return -1;return read(sock, remote, size) == size ? 0 : -1;
}/*** 轮询完成队列直到操作完成或超时* @return 成功返回0，失败返回-1*/
int poll_cq(struct ibv_cq *cq) {struct ibv_wc wc;unsigned long start = time(NULL) * 1000;while (time(NULL) * 1000 - start < MAX_POLL_CQ_TIMEOUT) {if (ibv_poll_cq(cq, 1, &wc) && wc.status == IBV_WC_SUCCESS) return 0;}return -1;
}/*** 提交发送工作请求（Work Request）* @param res 资源结构体指针* @param op 操作类型（IBV_WR_RDMA_WRITE/READ）*/
void post_send(struct resources *res, int op) {struct ibv_send_wr sr;struct ibv_sge sge;struct ibv_send_wr *bad_wr = NULL;memset(&sge, 0, sizeof(sge));sge.addr = (uintptr_t)res->buf;sge.length = res->mr->length;sge.lkey = res->mr->lkey;memset(&sr, 0, sizeof(sr));sr.next = NULL;sr.wr_id = 0;sr.sg_list = &sge;sr.num_sge = 1;sr.opcode = op;sr.send_flags = IBV_SEND_SIGNALED;sr.wr.rdma.remote_addr = res->remote_addr;sr.wr.rdma.rkey = res->remote_rkey;assert(0==ibv_post_send(res->qp, &sr, &bad_wr));
}/*----------------------------- 主函数 -----------------------------*/
int main(int argc, char **argv) {struct resources res = {0};char *remote_addr=0;//分配HOST或设备内存#ifdef USE_CPUres.buf = malloc(MSG_SIZE);memset(res.buf,0,MSG_SIZE);#endif    #ifdef USE_CUDAcuInit(0);CUdevice cuDevice;cuDeviceGet(&cuDevice, 0);CUcontext cuContext;assert(0 == cuCtxCreate(&cuContext, 0, cuDevice));assert(0 == cudaMallocHost(&res.buf, MSG_SIZE));struct cudaPointerAttributes cu_attrs;assert(0==cudaPointerGetAttributes(&cu_attrs,res.buf));unsigned cuflags = 1;assert(0==cuPointerSetAttribute(&cuflags, CU_POINTER_ATTRIBUTE_SYNC_MEMOPS,(CUdeviceptr)cu_attrs.devicePointer));assert(res.buf);cudaMemset(res.buf,0,MSG_SIZE);#endif    #ifdef USE_VACCint DevId=0;assert(0==vaccrtSetDevice(DevId));   assert(0==vaccrtMalloc(64<<10,MSG_SIZE,&res.buf));assert(0==vaccrtMemset(res.buf,0,MSG_SIZE)); int mem_handle=0;assert(0==vaccrtMemGetHandleForAddressRange(&mem_handle,res.buf, MSG_SIZE,VACCRT_MEM_RANGE_HANDLE_TYPE_DMA_BUF_FD,0));printf("mem_handle:%d ddr:%p\n",mem_handle,res.buf);#endifprintf("init done\n");// 参数解析：本地地址必须，设备名必须，远程地址可选config.local_addr = argv[1];    // 本地IP地址char *dev_name = argv[2];       // IB设备名（如mlx5_0）if(argc > 3) remote_addr = argv[3];  // 远程IP地址（客户端模式）/*=============== 阶段1：建立TCP连接 ===============*/// 用于交换RDMA连接参数（QPN、LID、内存地址等）    res.sock = sock_connect(config.local_addr, config.port,remote_addr);assert(res.sock>=0);/*=============== 阶段2：初始化IB资源 ===============*/// 1. 获取IB设备列表并打开指定设备    int num_devices=0;struct ibv_device * cur_dev=NULL;struct ibv_device **dev_list = ibv_get_device_list(&num_devices);for (int i = 0; i < num_devices; i++) {const char *name = ibv_get_device_name(dev_list[i]);printf("%d:%s\n",i,name);if (dev_name && strcmp(name, dev_name) == 0) {res.ctx = ibv_open_device(dev_list[i]);  cur_dev=dev_list[i];break;}}    assert(res.ctx);// 2. 查询IB端口属性（获取LID等信息）assert(0==ibv_query_port(res.ctx,1, &res.port_attr));// 3. 创建保护域（PD），用于管理资源权限res.pd = ibv_alloc_pd(res.ctx);assert(res.pd);// 4. 创建完成队列（CQ），用于接收操作完成通知res.cq = ibv_create_cq(res.ctx, 1, NULL, NULL, 0);assert(res.cq);// 5. 注册内存区域（MR），允许远程访问   int mr_flags=IBV_ACCESS_LOCAL_WRITE | IBV_ACCESS_REMOTE_WRITE | IBV_ACCESS_REMOTE_READ;#ifdef USE_CPUres.mr = ibv_reg_mr(res.pd, res.buf, MSG_SIZE,mr_flags );#endif#ifdef USE_VACCres.mr = ibv_reg_dmabuf_mr(res.pd,0,MSG_SIZE,(uint64_t)res.buf,mem_handle,mr_flags);#endif #ifdef USE_CUDAres.mr = ibv_reg_mr(res.pd, res.buf, MSG_SIZE,mr_flags);   #endifif(0==res.mr){printf("ibv_reg_mr %s (错误码=%d)\n", strerror(errno), errno);return -1;}// 6. 创建队列对（QP），RC（可靠连接）类型struct ibv_qp_init_attr qp_init_attr;memset(&qp_init_attr, 0, sizeof(qp_init_attr));qp_init_attr.qp_type = IBV_QPT_RC;qp_init_attr.sq_sig_all = 1;qp_init_attr.send_cq = res.cq;qp_init_attr.recv_cq = res.cq;qp_init_attr.cap.max_send_wr = 1;qp_init_attr.cap.max_recv_wr = 1;qp_init_attr.cap.max_send_sge = 1;qp_init_attr.cap.max_recv_sge = 1;res.qp = ibv_create_qp(res.pd, &qp_init_attr);assert(res.qp);/*=============== 阶段3：交换QP参数 ===============*/// 通过TCP交换双方的QP信息（地址、密钥等）    union ibv_gid my_gid;assert(0==ibv_query_gid(res.ctx, 1,0,&my_gid));// 本地参数打包struct { uint64_t addr;    // 内存地址uint16_t lid;     // 本地标识符uint32_t rkey;    // 内存访问密钥uint32_t qpn;     // QP编号uint8_t gid[16];  // 全局标识符（RoCEv2需要）} local = {(uintptr_t)res.buf,res.port_attr.lid,res.mr->rkey,res.qp->qp_num},remote;memcpy(local.gid, &my_gid, 16);// 参数交换（TCP同步）assert(0 == sock_sync(res.sock, sizeof(local), &local, &remote));// 保存远程参数res.remote_addr = remote.addr;res.remote_rkey = remote.rkey;printf("local: addr=%p, lkey=0x%x, rkey=0x%x\n", res.mr->addr, res.mr->lkey, res.mr->rkey);printf("remote: addr=%ld, rkey=0x%x\n", remote.addr, remote.rkey);/*=============== 阶段4：QP状态转换 ===============*/// QP需要经历三个状态变化：INIT -> RTR（准备接收） -> RTS（准备发送）// 1. INIT状态设置struct ibv_qp_attr attr;memset(&attr, 0, sizeof(attr));attr.qp_state = IBV_QPS_INIT;attr.port_num = 1;attr.pkey_index = 0;attr.qp_access_flags = IBV_ACCESS_LOCAL_WRITE | IBV_ACCESS_REMOTE_READ | IBV_ACCESS_REMOTE_WRITE;int flags = IBV_QP_STATE | IBV_QP_PKEY_INDEX | IBV_QP_PORT |IBV_QP_ACCESS_FLAGS;assert(0==ibv_modify_qp(res.qp, &attr, flags));// 2. RTR状态设置（准备接收）memset(&attr, 0, sizeof(attr));attr.qp_state = IBV_QPS_RTR;attr.path_mtu = IBV_MTU_256;attr.dest_qp_num = remote.qpn;attr.rq_psn = 0;attr.max_dest_rd_atomic = 1;attr.min_rnr_timer = 0x12;attr.ah_attr.is_global = 1;attr.ah_attr.dlid = remote.lid;attr.ah_attr.sl = 0;attr.ah_attr.src_path_bits = 0;attr.ah_attr.port_num = 1;attr.ah_attr.is_global = 1;attr.ah_attr.port_num = 1;// 如果是RoCEv2需要设置GRHattr.ah_attr.grh.flow_label = 0;attr.ah_attr.grh.hop_limit = 1;attr.ah_attr.grh.sgid_index = 0;attr.ah_attr.grh.traffic_class = 0;memcpy(&attr.ah_attr.grh.dgid, remote.gid, 16);flags = IBV_QP_STATE | IBV_QP_AV | IBV_QP_PATH_MTU | IBV_QP_DEST_QPN |IBV_QP_RQ_PSN | IBV_QP_MAX_DEST_RD_ATOMIC |IBV_QP_MIN_RNR_TIMER;assert(0==ibv_modify_qp(res.qp, &attr,flags));// 3. RTS状态设置（准备发送）memset(&attr, 0, sizeof(attr));attr.qp_state = IBV_QPS_RTS;attr.timeout = 0x12;attr.retry_cnt = 6;attr.rnr_retry = 0;attr.sq_psn = 0;attr.max_rd_atomic = 1;flags = IBV_QP_STATE | IBV_QP_TIMEOUT | IBV_QP_RETRY_CNT |IBV_QP_RNR_RETRY | IBV_QP_SQ_PSN | IBV_QP_MAX_QP_RD_ATOMIC;assert(0==ibv_modify_qp(res.qp, &attr,flags ));/*=============== 阶段5：数据传输 ===============*/if (remote_addr) {// 客户端模式，执行RDMA Write#ifdef USE_CPUstrcpy(res.buf, MSG);#endif#ifdef USE_CUDAassert(0==cudaMemcpy(res.buf, MSG, strlen(MSG), cudaMemcpyHostToDevice));#endif#ifdef USE_VACCassert(0==vaccrtMemcpy(MSG,strlen(MSG),res.buf,kHost2Ddr));  #endifpost_send(&res, IBV_WR_RDMA_WRITE);// 将本地数据写入远程内存} else {                               // 服务端模式，执行RDMA Readpost_send(&res, IBV_WR_RDMA_READ); // 从远程内存读取数据到本地}// 等待操作完成assert(0==poll_cq(res.cq));#ifdef USE_VACC{char buffer[MSG_SIZE];memset(buffer,0,MSG_SIZE);assert(0==vaccrtMemcpy(res.buf,strlen(MSG),buffer,kDdr2Host));  printf("Message: %s\n", buffer);}#endif#ifdef USE_CUDA{char buffer[MSG_SIZE];memset(buffer,0,MSG_SIZE);assert(0==cudaMemcpy(buffer,res.buf,strlen(MSG),cudaMemcpyDeviceToHost));  printf("Message: %s\n", buffer);}#endif#ifdef USE_CPUprintf("Message: %s\n", (const char*)res.buf);#endifprintf("end,Enter exit..\n");getchar(),getchar();/*=============== 阶段6：资源清理 ===============*/// 注意销毁顺序：QP -> MR -> CQ -> PD -> 上下文ibv_destroy_qp(res.qp);ibv_dereg_mr(res.mr);#ifdef USE_CUDAcudaFree(res.buf);#endif#ifdef USE_VACCvaccrtFree(res.buf);#endif#ifdef USE_CPUfree(res.buf);#endifibv_destroy_cq(res.cq);ibv_dealloc_pd(res.pd);ibv_close_device(res.ctx);close(res.sock);ibv_free_device_list(dev_list);return 0;
}

四、注意事项

• 1.ibv_reg_dmabuf_mr需要传入之前分配的设备地址,否则会出

  mlx5_0/1: QP 574 error: local protection error (0x3b 0x0 0x9d)
  

• 2.MAX_POLL_CQ_TIMEOUT值太小,会超时
• 3.cudaMalloc的内存在ibv_reg_mr时提示地址非法
• 4.收发完毕后才能释放资源