夜雨聆风最近在工作中遇到一个RDMA网络中使用perftest工具打流带宽异常问题，为定位该问题重现学习了一遍RDMA的原理和RDMA perftest的源码实现。
RDMA介绍
RDMA Perftest源码介绍
RDMA perftest是 Linux RDMA 社区（linux-rdma）提供的一套RDMA 微基准测试（Micro Benchmark）工具，用于评估 RDMA 网络和 RNIC（如 NVIDIA ConnectX 系列）性能。Perftest 通过调用RDMA Verbs API，直接对 RNIC 发起 RDMA 操作，测量不同通信模式下的：Latency（时延）、Bandwidth（带宽）、Message Rate（消息速率，Mpps）等。
以ib_write_bw为例解析RDMA perftest源码流程，完整实现参数解析→硬件初始化→两端握手协商→QP/MR 资源创建→RDMA 压测收发循环→性能统计输出→资源销毁全流程。ib_write_bw入口main在write_bw.c文件中。
intmain(int argc, char *argv[]){int ret_parser, i = 0, rc;struct ibv_device *ib_dev = NULL;struct pingpong_context ctx;//保存硬件资源容器（QP\MR\CQ等）struct pingpong_dest *my_dest,*rem_dest;//本地/远端QP链路信息struct perftest_parameters user_param; // 全局测试参数struct perftest_comm user_comm;//两端控制面通信上下文struct bw_report_data my_bw_rep, rem_bw_rep;int rdma_cm_flow_destroyed = 0;/* init default values to user's parameters */memset(&user_param,0,sizeof(struct perftest_parameters));memset(&user_comm,0,sizeof(struct perftest_comm));memset(&ctx,0,sizeof(struct pingpong_context));// 默认参数：RDMA WRITE、带宽测试、填充版本号user_param.verb = WRITE;user_param.tst = BW;ret_parser = parser(&user_param,argv,argc);// 解析命令行参数if((user_param.connection_type == DC || user_param.use_xrc)&& user_param.duplex) {user_param.num_of_qps *= 2; // XRC/DC双向QP数量修正}// 查找 RoCE/IB 网卡设备 + CPU 亲和绑定ib_dev = ctx_find_dev(&user_param.ib_devname);if (set_process_affinity(ib_dev->ibdev_path, &user_param)){}// 创建libibvervs 上下文，校验参数是否被网卡支持ctx.context = ctx_open_device(ib_dev, &user_param);verify_params_with_device_context(ctx.context, &user_param);check_link(ctx.context,&user_param);//校验网卡速率// 创建TCP socket用来交换两端控制面信息create_comm_struct(&user_comm,&user_param);establish_connection(&user_comm); // SERVER阻塞等待客户端TCP连接// 分配本地/远端QP链路信息数组MAIN_ALLOC(my_dest , struct pingpong_dest , user_param.num_of_qps , free_rdma_params);memset(my_dest, 0, sizeof(struct pingpong_dest)*user_param.num_of_qps);MAIN_ALLOC(rem_dest , struct pingpong_dest , user_param.num_of_qps , free_my_dest);memset(rem_dest, 0, sizeof(struct pingpong_dest)*user_param.num_of_qps);//分配全局CTX资源（内存池、PD\QP\MR）alloc_ctx(&ctx,&user_param);// 通过 TCP socket 同步两端测试配置negotiate_params(&ctx, &user_comm, &user_param);//初始化链路（两条分支：RDMA_CM和传统直连QP）if (user_param.work_rdma_cm == ON) { // 由rdma_cm管理QP建立、路由解析// 使用 rdma_cm 库自动完成 GID 路由、QP 连接建立，适配 RoCE 多网卡场景；rc = create_rdma_cm_connection(&ctx, &user_param, &user_comm, my_dest, rem_dest);} else { //手动创建QP\PD\MR\CQ资源if (ctx_init(&ctx, &user_param)) {}}// 数据校验模块初始化 开启--data-validation时使用if (user_param.data_validation &&(user_param.machine == SERVER || user_param.duplex) &&data_validation_init(&ctx, &user_param)) {}// 填充本地QP完整链路信息，根据mr_per_qp/共享mr分别计算偏移if (set_up_connection(&ctx,&user_param,my_dest)) {}// 两端QP握手交换链路信息ctx_print_test_info(&user_param);for (i=0; i < user_param.num_of_qps; i++) {if(ctx_hand_shake(&user_comm,&my_dest[i],&rem_dest[i])) {}}if (user_param.work_rdma_cm == OFF) {// QP状态机推进：INIT->RTR->RYSif (ctx_connect(&ctx,rem_dest,&user_param,my_dest)) {}}if (user_param.connection_type == DC){if (set_up_connection(&ctx, &user_param, my_dest)){}}// 打印本地和对端QP信息：LID/QPN/PSN/RKey/LKey/VAddr/GIDfor (i=0; i < user_param.num_of_qps; i++)ctx_print_pingpong_data(&my_dest[i],&user_comm);user_comm.rdma_params->side = REMOTE;for (i=0; i < user_param.num_of_qps; i++) {if (ctx_hand_shake(&user_comm,&my_dest[i],&rem_dest[i])) {}ctx_print_pingpong_data(&rem_dest[i],&user_comm);}if (user_param.use_event) {// CQ事件通知注册，不开启则轮询poll CQ队列if (ibv_req_notify_cq(ctx.send_cq, 0)) {}if (ibv_req_notify_cq(ctx.recv_cq, 0)) {}}/* An additional handshake is required after moving qp to RTR. */if (ctx_hand_shake(&user_comm,&my_dest[0],&rem_dest[0])) {}// server 半双工write特殊分支。仅等待客户端压测，无发包if (user_param.machine == SERVER &&user_param.verb == WRITE && !user_param.duplex)// 三大压测主循环分支：// 分支一：RUN_ALL多报文长度遍历（1B~8M 的2的n次幂）if (user_param.test_method == RUN_ALL){for(int i=0;i<24;i++){if(client)ctx_set_send_wqes(&ctx,&user_param);if(server)ctx_set_recv_wqes(&ctx,&user_param);if (user_param.duplex && user_param.verb == WRITE_IMM)run_iter_bi(&ctx,&user_param);// 标准通用带宽压测客户端else if((user_param.machine == CLIENT || user_param.verb != WRITE_IMM)//run_iter_bw_dv是带数据校验的压测客户端；run_iter_bw:普通压测客户端(user_param.data_validation ? run_iter_bw_dv : run_iter_bw)(&ctx,&user_param);else if(user_param.machine == SERVER)run_iter_bw_server(&ctx,&user_param);}}// 分支二：RUN_REGULAR 固定报文长度单次压测elseif (user_param.test_method== RUN_REGULAR){if(client)ctx_set_send_wqes(&ctx,&user_param);if(server)ctx_set_recv_wqes(&ctx,&user_param);}// 分支三： RUN_INFINITEFY 无限循环压测elseif (user_param.test_method== RUN_INFINITELY) {run_iter_bw_infinitely(&ctx,&user_param)；}// 校验带宽/消息速率是否达到用户设定最低阈值if (!user_param.is_bw_limit_passed && (user_param.is_limit_bw == ON ) ) {}// 关闭QP链路，推进QP到ERROR状态释放硬件资源ctx_close_connection(&user_comm,&my_dest[0],&rem_dest[0]);// 多层goto统一释放资源}
pingpong_context ctx 批量动态分配所有 RDMA 运行时数组、硬件资源句柄数组、计时统计数组、收发 SGE/WQE 缓冲区，统一计算全局缓冲区总长度 ctx->buff_size，最后初始化内存抽象层 ctx->memory。intalloc_ctx(structpingpong_context*ctx,structperftest_parameters*user_param){// 分配QP、MR句柄、缓冲区起始地址虚拟地址ALLOC(ctx->qp, struct ibv_qp*, user_param->num_of_qps);memset(ctx->qp, 0, user_param->num_of_qps*sizeof (structibv_qp*));ALLOC(ctx->mr, struct ibv_mr*, user_param->num_of_qps);ALLOC(ctx->buf, void*, user_param->num_of_qps);// 带宽测试 && （客户端||双向模式）if ((user_param->tst==BW||user_param->tst==LAT_BY_BW) &&(user_param->machine==CLIENT||user_param->duplex)) {// 本地缓冲区VA数组，远端发布VAdd数组，发送计数，完成计数ALLOC(ctx->my_addr,uint64_t,user_param->num_of_qps);ALLOC(ctx->rem_addr,uint64_t,user_param->num_of_qps);ALLOC(ctx->scnt,uint64_t,user_param->num_of_qps);ALLOC(ctx->ccnt,uint64_t,user_param->num_of_qps);} elseif ((user_param->tst==BW||user_param->tst==LAT_BY_BW)&&has_recv_comp(user_param->verb) &&user_param->machine==SERVER) {// 带宽测试，服务端仅需本地缓冲区地址，无需远端地址、发送计数ALLOC(ctx->my_addr, uint64_t, user_param->num_of_qps);ALLOC(user_param->tcompleted, cycles_t, 1);} elseif (user_param->tst==FS_RATE&&user_param->test_type==ITERATIONS) {// 固定迭代吞吐测试ALLOC(user_param->tcompleted, cycles_t, tarr_size);memset(user_param->tcompleted, 0, sizeof(cycles_t) * tarr_size);}// 发送侧 SGE/WQE、远端 QP 编号、地址句柄分配（CLIENT / 双向 / 时延测试）if (user_param->machine==CLIENT||user_param->tst==LAT||user_param->duplex) {ALLOC(ctx->current_send_buffer_offset, int, user_param->num_of_qps);ALLOC(ctx->current_remote_recv_offset, int, user_param->num_of_qps);memset(ctx->current_send_buffer_offset, 0, sizeof(int) *user_param->num_of_qps);memset(ctx->current_remote_recv_offset, 0, sizeof(int) *user_param->num_of_qps);if (user_param->data_validation) { // 开启校验，SGE区分长度if (user_param->verb==READ) { // READ：每个校验分片独立SGE，数量=分片数×QP数......} else { //WRITE：固定3条SGE（数据、校验、尾标记）ALLOC(ctx->sge_list, struct ibv_sge, 3);}// 数据+原子操作两套WRALLOC(ctx->wr, struct ibv_send_wr, user_param->num_of_qps*2);} else { // 无校验：SGE/WR数量=QP数×单次投递报文数post_listALLOC(ctx->sge_list, struct ibv_sge,user_param->num_of_qps*user_param->post_list);ALLOC(ctx->wr, struct ibv_send_wr, user_param->num_of_qps*user_param->post_list);}// 远端QP编号数组ALLOC(ctx->rem_qpn, uint32_t, user_param->num_of_qps);// 无连接模式（UD/DC/SRD） 分配地址句柄AHif (((user_param->verb==SEND||user_param->verb==SEND_IMM) &&user_param->connection_type==UD) ||user_param->connection_type==DC||user_param->connection_type==SRD) {ALLOC(ctx->ah, struct ibv_ah*, user_param->num_of_qps);}}// 接收侧SGE/WR/接收地址数组分配if (has_recv_comp(user_param->verb) && (user_param->tst==LAT||user_param->machine==SERVER||user_param->duplex)) {ALLOC(ctx->recv_sge_list, struct ibv_sge,user_param->num_of_qps*user_param->recv_post_list);ALLOC(ctx->rwr, struct ibv_recv_wr,user_param->num_of_qps*user_param->recv_post_list);ALLOC(ctx->rx_buffer_addr, uint64_t, user_param->num_of_qps);}// 全局缓冲区尺寸计算ctx->size=user_param->size;num_of_qps_factor= (user_param->mr_per_qp) ?1:user_param->num_of_qps;// 计算公式拆解：// 1. BUFF_SIZE：报文长度向上对齐cycle_buffer单块尺寸// 2. INC：再向上对齐CPU cache_line_size，消除伪共享// 3. ×2：一块缓冲区拆分为本地发送区、远端接收区（偏移 (num_qps+i)*S 的来源）// 4. ×num_of_qps_factor：共享MR=总QP数，独立MR=1// 5. ×flows：单QP多流并发倍数ctx->buff_size=INC(BUFF_SIZE(ctx->size, ctx->cycle_buffer),ctx->cache_line_size) *2*num_of_qps_factor*user_param->flows;// 单QP发送缓冲区总长度ctx->send_qp_buff_size=ctx->buff_size/num_of_qps_factor/2;// 单流分片长度ctx->flow_buff_size=ctx->send_qp_buff_size/user_param->flows;user_param->buff_size=ctx->buff_size; //同步给全局参数if (user_param->connection_type==UD)// 无连接模式额外加缓存行防止越界ctx->buff_size+=ctx->cache_line_size;if (user_param->data_validation) { //数据校验模式重算缓冲区布局ctx->payload_size=ctx->size;structvalidation_buffer_layoutlayout;//开启数据校验后，内存不再简单均分，会划分独立校验分片、尾标记区域，覆盖之前的//buff_size，对应set_up_connection 中 WRITE 模式不规则 VAddr 偏移分支。if (compute_validation_layout(ctx->payload_size,user_param->validation_chunk_size,user_param->num_of_qps,user_param->validation_chunks_per_qp,(user_param->verb==READ),&layout) !=0) {}ctx->tail_markers_offset=layout.markers_offset;ctx->recv_slots_offset=layout.recv_slots_offset;ctx->atomic_returns_offset=layout.atomic_returns_offset;ctx->buff_size=layout.total_size;}// 初始化内存抽 ctx->memoryctx->memory=user_param->memory_create(user_param);returnSUCCESS;}
intctx_init(structpingpong_context*ctx, structperftest_parameters*user_param){/* 创建CQ完成事件通道. 开启事件等待模式，时间唤醒进程代替CPU轮询*/if (user_param->use_event) {ctx->send_channel=ibv_create_comp_channel(ctx->context);ctx->recv_channel=ibv_create_comp_channel(ctx->context);}/*分配PD保护域 PD：RDMA基础隔离单元，所有MR/QP绑定统一PD才能相互访问*/ctx->pd=ibv_alloc_pd(ctx->context);// 初始化内存抽相层 ，调用maca_memory_allocate_bufferif (ctx->memory->init(ctx->memory)) {}// 批量创建MR内存（核心调用create_mr）if (create_mr(ctx, user_param)) {}// 创建CQ发送/接收完成队列if (create_cqs(ctx, user_param)) {}//创建SRQ共享接收队列（DC/LAT/双向测试）if (user_param->use_srq&&user_param->connection_type==DC&&(user_param->tst==LAT||user_param->machine==SERVER||user_param->duplex==ON)){structibv_srq_init_attr_exattr;memset(&attr, 0, sizeof(attr));attr.comp_mask= IBV_SRQ_INIT_ATTR_TYPE | IBV_SRQ_INIT_ATTR_PD;attr.attr.max_wr=user_param->rx_depth;attr.attr.max_sge=1;attr.pd=ctx->pad;attr.srq_type= IBV_SRQT_BASIC;ctx->srq=ibv_create_srq_ex(ctx->context, &attr);if (!ctx->srq) {fprintf(stderr, "Couldn't create SRQ\n");gotoxrc_srq;}}// 批量创建QP队列for (i=0; i<user_param->num_of_qps; i++) {if (create_qp_main(ctx, user_param, i)) {}qp_index++;}returnSUCCESS;}
set_up_connection批量填充本地每条 QP 的完整远端通信描述体 pingpong_dest my_dest[]，生成两端握手必须的全套链路元数据：LID、GID、QPN、随机 PSN、MR 远端密钥 RKey、对外发布的远端虚拟地址 VAddr、SRQ 共享接收队列号 SRQN。 填充完成后，通过 TCP 控制面与对端交换 my_dest/rem_dest。intset_up_connection(structpingpong_context*ctx,structperftest_parameters*user_param,structpingpong_dest*my_dest){// DC/XRC架构收发QP分离，双向/时延测试时物理QP分为发送、接收各一半if ((user_param->connection_type==DC||user_param->use_xrc) && (user_param->duplex||user_param->tst==LAT)) {num_of_qps/=2;num_of_qps_per_port=num_of_qps/2;}if (user_param->gid_index!=-1) {// 读取端口状态、MTU、速率if (ibv_query_port(ctx->context, user_param->ib_port, &attr))}// 双卡 RoCE 场景，缓存第二块网卡 GID 到temp_gid2，循环分配 QP 时绑定对应网卡标识。if (user_param->dualport==ON) { //双端口第二张网卡IB_PORT2 GID获取if (user_param->gid_index2!=-1) {if (ibv_query_port(ctx->context, user_param->ib_port2, &attr))return0;}}// 循环填充每个QP的QPN\PSN\RKEY三大硬件标识for (i=0; i<user_param->num_of_qps; i++) {my_dest[i].qpn =ctx->qp[i]->qp_num;my_dest[i].psn =lrand48() &0xffffff;my_dest[i].rkey=user_param->use_null_mr?ctx->null_mr->lkey:ctx->mr[i]->rkey; */if (user_param->mr_per_qp) // 对外发布远端vaddrmy_dest[i].vaddr= (uintptr_t)ctx->buf[i] +BUFF_SIZE(ctx->size,ctx->cycle_buffer);elseif (user_param->data_validation==ON) {if (user_param->verb==READ) {/* ib_read_bw: 所有QP统一度内存起始基地址 */my_dest[i].vaddr= (uintptr_t)ctx->buf[0];} else {/* ib_WRITE+bw: 每条QP分配独立校验分片，偏移不规则 */my_dest[i].vaddr= (uintptr_t)ctx->buf[0] +ctx->recv_slots_offset+ i * (user_param->validation_chunks_per_qp*user_param->validation_chunk_size*ctx->payload_size);}/* 校验结束标记地址，同步数据校验完成状态 */my_dest[i].tail_markers_vaddr= (uintptr_t)ctx->buf[0] +ctx->tail_markers_offset;} else // 共享MR、无数据校验my_dest[i].vaddr= (uintptr_t)ctx->buf[0] + (user_param->num_of_qps+ i)*BUFF_SIZE(ctx->size,ctx->cycle_buffer);}return0;}
voidctx_set_send_wqes(struct pingpong_context *ctx,struct perftest_parameters *user_param,struct pingpong_dest *rem_dest){if (user_param->data_validation && rem_dest !=NULL) {if (user_param->verb == READ)ctx_set_send_wqes_data_val_read(ctx, user_param, rem_dest);elsectx_set_send_wqes_data_val_write(ctx, user_param, rem_dest);} else {ctx_set_send_reg_wqes(ctx,user_param,rem_dest);}}
ctx_set_send_reg_wqes() 是 perftest 最核心的发送初始化函数。它根据测试参数（READ/WRITE/SEND/ATOMIC、BW/LAT、post_list等），预先构造好所有 ibv_sge 和 ibv_send_wr，供 ibv_post_send() 直接提交到 RNIC。ctx_set_send_wqes_data_val_read/ctx_set_send_wqes_data_val_write带数据校验场景专用初始化函数。三个函数区别如下：ctx_set_send_reg_wqes：通用带宽 / 延迟压测，无数据校验ctx_set_send_wqes_data_val_read：校验读流程（READ 拉远端数据）ctx_set_send_wqes_data_val_write：校验写流程（WRITE 推送数据到远端）
run_iter_bw：单向发送客户端（标准吞吐，无校验、无接收处理） run_iter_bw_dv：单向发送客户端（带数据完整性校验 DV） run_iter_bw_server：纯接收服务端（只收、补发 Recv、回 credit 流控，不发业务包） run_iter_bi：Bidirectional 双向一体化收发（单线程同时发包 + 收包，支持 Client/Server）
staticinlineintpost_send_method(struct pingpong_context *ctx, intindex,struct perftest_parameters *user_param){#ifdef HAVE_IBV_WR_APIif (!user_param->use_old_post_send)return (*ctx->new_post_send_work_request_func_pointer)(ctx, index, user_param);#endifstruct ibv_send_wr *bad_wr =NULL;return ibv_post_send(ctx->qp[index], &ctx->wr[index*user_param->post_list], &bad_wr);}
ibv_send_wr（发送工作请求 WR）批量下发到 QP 的发送队列，网卡硬件异步执行 RDMA 操作：SEND / WRITE / READ / ATOMIC。https://github.com/linux-rdma/perftest
