引言：高性能计算的瓶颈与RDMA的崛起
在现代分布式系统和高性能计算（HPC）领域，数据的高效传输与访问是决定系统整体性能的关键因素。传统基于TCP/IP协议栈的网络通信，尽管通用且可靠，但在面对极低延迟和极高带宽需求时，其固有的开销日益成为瓶颈。这些开销主要源于：
- CPU参与度高：数据在用户空间与内核空间之间频繁拷贝，每次拷贝都需要CPU介入。
- 协议栈深度：从应用层到物理层，数据需要经过多层协议处理，引入大量软件开销。
- 内存拷贝：数据从应用缓冲区到内核缓冲区，再到网卡缓冲区，通常需要多次内存拷贝，增加了延迟并消耗CPU周期。
为了突破这些限制，远程直接内存访问（RDMA）技术应运而生。RDMA允许网络适配器（通常称为Host Channel Adapter, HCA）直接读写远程机器的内存，而无需远程CPU的参与。这种“零拷贝”（Zero-copy）和“CPU卸载”（CPU Offload）的特性，使得RDMA能够显著降低通信延迟，提高吞吐量，并大幅减少CPU的负载，将CPU资源解放出来用于实际的计算任务。
本文将深入探讨如何在C++中利用RDMA的单边操作（One-sided Operations）实现一个跨节点、零拷贝的内存池。我们将从RDMA的核心概念出发，逐步讲解C++中RDMA Verbs API的编程基础，最终构建一个能够进行高效远程读写的内存池框架。
RDMA核心概念：通往硬件加速之路
RDMA技术的核心在于其绕过操作系统内核和CPU的直接内存访问能力。理解以下关键概念是进行RDMA编程的基础：
1. RDMA Verbs API概述
RDMA Verbs API是操作系统提供给用户空间应用程序的接口，用于直接与HCA交互。它是一组C语言函数，类似于POSIX API，但专门用于管理RDMA设备、创建通信通道、注册内存以及提交RDMA操作。所有RDMA设备（如InfiniBand、RoCEv2、iWARP）都通过这套统一的API进行编程。
2. 保护域 (Protection Domain, PD)
保护域是RDMA资源（如内存区域、队列对）的管理单元。所有属于同一个PD的资源可以相互访问。PD的主要作用是提供一种隔离和安全机制，防止未经授权的内存访问。在进行任何RDMA操作之前，必须先创建一个PD。
3. 内存区域 (Memory Region, MR) 及其注册 (Registration)
RDMA操作只能针对已注册的内存区域进行。内存注册是将一段用户空间的虚拟内存地址映射到HCA可以访问的物理地址，并告知HCA这段内存的权限（读、写、原子操作）。注册成功后，HCA会返回一个本地键（Local Key, lkey）和一个远程键（Remote Key, rkey）。
lkey用于本地发起RDMA操作时标识源内存。rkey用于远程发起RDMA操作时标识目标内存。
内存注册是RDMA零拷贝的关键一步，因为它将应用程序缓冲区直接暴露给HCA，避免了数据在系统缓冲区和用户缓冲区之间的拷贝。由于内存注册通常涉及页锁定（page pinning），以防止操作系统将页面换出，因此它是一个相对昂贵的操作。在设计内存池时，通常会注册大块内存，然后在其中进行细粒度管理。
4. 队列对 (Queue Pair, QP)
队列对是RDMA通信的基本单元，类似于传统网络中的socket。每个QP包含一个发送队列（Send Queue, SQ）和一个接收队列（Receive Queue, RQ）。
- SQ 用于提交由本地HCA发起的RDMA操作（如RDMA Read、RDMA Write、Send）。
- RQ 用于接收由远程HCA发起的RDMA操作（如Recv）。
单边操作（RDMA Read/Write/Atomic）只使用SQ，因为它们不需要远程CPU的参与或远程接收队列的缓冲。双边操作（Send/Recv）则需要SQ和RQ协同工作。
QP有多种状态，它们之间的转换是严格有序的，常见的状态包括：
- RESET: 初始状态。
- INIT: 配置QP的基本属性。
- RTR (Ready to Receive): QP准备好接收传入连接。
- RTS (Ready to Send): QP准备好发送数据。
5. 完成队列 (Completion Queue, CQ)
完成队列用于接收RDMA操作的完成通知。当一个RDMA操作（无论是发送还是接收）完成后，HCA会将一个工作完成（Work Completion, WC）对象放入相应的CQ。应用程序通过轮询（ibv_poll_cq）或事件通知机制来检查WC，从而得知操作是否成功完成。
6. 工作请求 (Work Request, WR) 与工作完成 (Work Completion, WC)
- 工作请求 (WR)：应用程序通过构建一个WR结构体并将其提交到SQ或RQ来发起RDMA操作。WR包含了操作类型、源/目标地址、长度、
lkey/rkey等信息。 - 工作完成 (WC)：当一个WR成功或失败完成时，HCA会生成一个WC并将其放入CQ。WC包含了操作的状态、操作ID、字节数等信息，用于应用程序进行错误检查和流程控制。
7. 散列/聚集列表 (Scatter/Gather List, SGL)
SGL允许一个RDMA操作从非连续的内存区域读取数据（gather）或将数据写入非连续的内存区域（scatter）。它是一个包含多个ibv_sge（Scatter/Gather Entry）结构体的数组，每个ibv_sge描述了一个内存段的地址、长度和lkey。这使得RDMA操作更加灵活，能够处理复杂的内存布局。
RDMA单边操作：无CPU参与的魔法
RDMA单边操作是RDMA技术最引人注目的特性之一。与传统的双边通信（如TCP Send/Recv，或RDMA Send/Recv）不同，单边操作允许一个节点（发起方）直接读写另一个节点（目标方）的内存，而无需目标节点的CPU参与。这极大地简化了编程模型，降低了通信开销，并实现了真正的零拷贝。
单边操作的本质：Read, Write, Atomic
RDMA提供了三种主要的单边操作：
- RDMA Read：发起方HCA直接从目标方HCA控制的内存区域读取数据到发起方的内存区域。
- RDMA Write：发起方HCA直接将数据从发起方的内存区域写入到目标方的内存区域。
- RDMA Atomic：发起方HCA在目标方内存区域执行原子操作（如Fetch&Add或Compare&Swap）。这对于实现分布式锁、计数器或实现内存池的分配/回收逻辑非常有用。
与双边操作 (Send/Recv) 的对比
| 特性 | RDMA 单边操作 (Read/Write/Atomic) | RDMA 双边操作 (Send/Recv) |
|---|---|---|
| CPU参与 | 目标端CPU零参与 | 目标端CPU需要参与接收操作 (将数据从RQ复制到应用缓冲区) |
| 内存拷贝 | 零拷贝 (HCA直接DMA) | 目标端可能需要拷贝 (从RQ到应用缓冲区) |
| 编程模型 | 简单，类似内存访问，但需要目标端MR信息 | 复杂，需要匹配的Send和Recv操作，类似传统socket |
| 延迟 | 极低，通常在微秒级别 | 略高于单边操作，但仍远低于TCP/IP |
| 使用场景 | 远程内存访问、分布式缓存、共享内存池 | 消息传递、流式数据传输 |
零拷贝的实现机制：DMA引擎直接操作内存
RDMA单边操作实现零拷贝的核心机制是直接内存访问（DMA）。当一个节点发起RDMA Read或Write操作时：
- 发起方应用程序提交一个包含源地址、目标地址、长度和
rkey的WR到本地HCA的SQ。 - 本地HCA验证WR的合法性，然后利用其DMA引擎直接从本地注册的内存区域读取数据（对于Write）或将数据写入本地注册的内存区域（对于Read）。
- 本地HCA通过网络将RDMA请求发送给目标HCA。这个请求包含了要访问的远程内存的
rkey和偏移量。 - 目标HCA接收到RDMA请求后，验证
rkey和访问权限。如果合法，目标HCA的DMA引擎将直接从其控制的内存区域读取数据（对于Read）或将数据写入其控制的内存区域（对于Write），绕过了目标节点的CPU和操作系统内核。 - 数据传输完成后，发起方HCA在本地CQ中放置一个WC，通知应用程序操作完成。
整个过程中，数据只在源内存和目标内存之间传输一次，没有任何中间缓冲区拷贝，也无需目标CPU进行数据处理。
C++ RDMA Verbs API编程基础
在C++中进行RDMA编程，主要是通过调用C风格的ibv_*函数来完成。为了提高代码的可读性和可维护性，我们通常会对这些函数进行封装，构建面向对象的RDMA上下文和连接管理类。
环境设置与库引用
首先，需要确保系统上安装了RDMA驱动和库（如libibverbs和librdmacm）。在编译C++程序时，需要链接这些库：
g++ your_program.cpp -o your_program -libverbs -lrdmacm核心结构体与函数概览
| 类别 | 结构体/函数示例 | 描述 |
|---|---|---|
| 设备 | ibv_device, ibv_get_device_list, ibv_open_device |
获取和打开RDMA设备 |
| 上下文 | ibv_context |
RDMA设备的抽象，包含设备能力、端口信息等 |
| 保护域 | ibv_pd, ibv_alloc_pd, ibv_dealloc_pd |
创建和释放保护域 |
| 内存区 | ibv_mr, ibv_reg_mr, ibv_dereg_mr |
注册和注销内存区域，获取lkey/rkey |
| 完成队列 | ibv_cq, ibv_create_cq, ibv_destroy_cq, ibv_poll_cq |
创建、销毁和轮询完成队列，获取工作完成 |
| 队列对 | ibv_qp, ibv_create_qp, ibv_destroy_qp, ibv_modify_qp |
创建、销毁和修改队列对状态 |
| 工作请求 | ibv_send_wr, ibv_sge, ibv_post_send |
构建工作请求，提交到发送队列 |
| 地址解析 | rdma_cm_id, rdma_create_id, rdma_resolve_addr, rdma_resolve_route |
用于RDMA通信管理器（RDMA CM）的地址解析和路由发现（主要用于RoCE/iWARP） |
| 连接管理 | rdma_listen, rdma_connect, rdma_accept |
用于RDMA CM的连接建立和管理 |
C++封装示例：RdmaContext 和 RdmaMemoryRegion
为了更好地管理RDMA资源，我们可以创建一些C++类来封装这些C API。
#include <infiniband/verbs.h>
#include <rdma/rdma_cma.h> // For RDMA CM, especially useful for RoCE/iWARP
#include <vector>
#include <string>
#include <stdexcept>
#include <iostream>
#include <memory>
#include <cstring> // For memset, memcpy
// 辅助函数：检查RDMA操作结果
void check_rdma_error(int ret, const std::string& msg) {
if (ret) {
throw std::runtime_error(msg + ": " + std::string(strerror(errno)));
}
}
// -------------------------------------------------------------------------
// RdmaContext: 封装RDMA设备、PD、CQ的初始化与管理
// -------------------------------------------------------------------------
class RdmaContext {
public:
RdmaContext(const std::string& dev_name = "") {
init_context(dev_name);
}
~RdmaContext() {
if (cq_) {
check_rdma_error(ibv_destroy_cq(cq_), "Failed to destroy CQ");
}
if (pd_) {
check_rdma_error(ibv_dealloc_pd(pd_), "Failed to deallocate PD");
}
if (ctx_) {
check_rdma_error(ibv_close_device(ctx_), "Failed to close device");
}
std::cout << "RDMA Context destroyed." << std::endl;
}
ibv_pd* get_pd() const { return pd_; }
ibv_cq* get_cq() const { return cq_; }
ibv_context* get_ctx() const { return ctx_; }
private:
ibv_context* ctx_ = nullptr;
ibv_pd* pd_ = nullptr;
ibv_cq* cq_ = nullptr;
void init_context(const std::string& dev_name) {
ibv_device** dev_list = nullptr;
int num_devices = 0;
// 1. 获取RDMA设备列表
dev_list = ibv_get_device_list(&num_devices);
if (!dev_list || num_devices == 0) {
throw std::runtime_error("No RDMA devices found.");
}
ibv_device* selected_dev = nullptr;
if (dev_name.empty()) {
selected_dev = dev_list[0]; // 默认选择第一个设备
std::cout << "Using default RDMA device: " << selected_dev->name << std::endl;
} else {
for (int i = 0; i < num_devices; ++i) {
if (std::string(dev_list[i]->name) == dev_name) {
selected_dev = dev_list[i];
std::cout << "Using specified RDMA device: " << selected_dev->name << std::endl;
break;
}
}
if (!selected_dev) {
ibv_free_device_list(dev_list);
throw std::runtime_error("Specified RDMA device not found: " + dev_name);
}
}
// 2. 打开RDMA设备
ctx_ = ibv_open_device(selected_dev);
check_rdma_error(!ctx_, "Failed to open RDMA device");
// 3. 释放设备列表
ibv_free_device_list(dev_list);
// 4. 分配保护域 (PD)
pd_ = ibv_alloc_pd(ctx_);
check_rdma_error(!pd_, "Failed to allocate PD");
// 5. 创建完成队列 (CQ)
// 这里的256是CQ的大小，即可以容纳多少个工作完成事件。根据并发操作数调整。
cq_ = ibv_create_cq(ctx_, 256, nullptr, nullptr, 0);
check_rdma_error(!cq_, "Failed to create CQ");
std::cout << "RDMA Context initialized successfully." << std::endl;
}
};
// -------------------------------------------------------------------------
// RdmaMemoryRegion: 封装内存区域的注册与注销
// -------------------------------------------------------------------------
class RdmaMemoryRegion {
public:
RdmaMemoryRegion(RdmaContext& ctx, void* addr, size_t length)
: ctx_(ctx), addr_(addr), length_(length) {
register_mr();
}
~RdmaMemoryRegion() {
if (mr_) {
check_rdma_error(ibv_dereg_mr(mr_), "Failed to deregister MR");
}
std::cout << "Memory Region deregistered." << std::endl;
}
ibv_mr* get_mr() const { return mr_; }
void* get_addr() const { return addr_; }
size_t get_length() const { return length_; }
private:
RdmaContext& ctx_;
void* addr_ = nullptr;
size_t length_ = 0;
ibv_mr* mr_ = nullptr;
void register_mr() {
// IBV_ACCESS_LOCAL_WRITE: 允许本地HCA写入此MR (必要)
// IBV_ACCESS_REMOTE_WRITE: 允许远程HCA写入此MR
// IBV_ACCESS_REMOTE_READ: 允许远程HCA读取此MR
// IBV_ACCESS_REMOTE_ATOMIC: 允许远程HCA对此MR执行原子操作
int access_flags = IBV_ACCESS_LOCAL_WRITE | IBV_ACCESS_REMOTE_READ | IBV_ACCESS_REMOTE_WRITE | IBV_ACCESS_REMOTE_ATOMIC;
mr_ = ibv_reg_mr(ctx_.get_pd(), addr_, length_, access_flags);
check_rdma_error(!mr_, "Failed to register memory region");
std::cout << "Memory Region registered. Addr: " << addr_ << ", Length: " << length_
<< ", LKey: " << mr_->lkey << ", RKey: " << mr_->rkey << std::endl;
}
};QP设置与状态转换
QP的设置相对复杂，因为它涉及到与远程QP的连接。对于RoCE或iWARP，librdmacm库可以简化连接管理。
QP的状态转换如下：
- RESET -> INIT: 创建QP后，它处于RESET状态。调用
ibv_modify_qp将其转换为INIT状态，并设置PD、CQ、端口号等基本属性。 - INIT -> RTR (Ready To Receive): 在INIT状态下，QP可以进入RTR状态。这时需要提供远程QP的连接信息，包括远程QP号、远程HCA的GID（全局ID）或IP地址。
- RTR -> RTS (Ready To Send): 在RTR状态下，QP可以进入RTS状态。这时需要设置本地QP的连接信息，如本地QP号、本地HCA的GID。一旦进入RTS状态，QP就可以发送和接收数据了。
对于QP的连接，通常有两种方式：
- 手动连接：通过
ibv_modify_qp系列函数手动设置远程QP信息，通常用于InfiniBand原生的GID/LID交换。 - RDMA CM连接：通过
librdmacm库提供的rdma_cm_id和rdma_connect/rdma_listen/rdma_accept等函数进行连接，它会自动处理地址解析和QP状态转换，尤其适用于RoCE/iWARP。
在构建跨节点内存池时，我们通常会利用RDMA CM来建立初始连接并交换QP信息，因为它更为便捷。
构建跨节点零拷贝内存池
现在我们结合RDMA单边操作和C++封装，来设计并实现一个跨节点的零拷贝内存池。
架构设计理念
一个跨节点内存池的核心思想是，每个节点都贡献一部分本地内存，并将其注册为RDMA MR。然后，这些MR的信息（如rkey、地址、长度）通过某种机制（通常是TCP/IP辅助通道）共享给其他节点。其他节点在需要访问远程数据时，可以直接使用这些rkey和地址发起RDMA单边操作。
主要组件：
- MemoryPoolServer: 在每个节点上运行，负责：
- 分配并注册大块本地内存。
- 监听来自客户端的连接请求。
- 通过辅助通道（如TCP）与客户端交换本地MR信息和QP连接信息。
- MemoryPoolClient: 应用程序使用此客户端接口来：
- 连接到远程MemoryPoolServer。
- 接收并存储远程节点的MR信息。
- 封装RDMA单边操作，提供类似于本地内存访问的API（
read_remote、write_remote、atomic_cas_remote）。
- 元数据交换通道: 独立的TCP/IP连接，用于在RDMA连接建立前交换必要的元数据，如HCA的GID、端口、QP号以及注册内存的
rkey、地址和长度。
远程内存句柄 (Remote Memory Handle)
为了抽象远程内存，我们可以定义一个结构体来表示远程内存段：
struct RemoteMemInfo {
uint64_t remote_addr; // 远程内存的起始地址
uint32_t rkey; // 远程内存的RKey
size_t length; // 远程内存的长度
uint32_t node_id; // 标识哪个节点拥有这段内存
};客户端会维护一个映射，将NodeID与该节点注册的RemoteMemInfo关联起来。
内存分配与释放策略
为了实现高效的跨节点内存池，我们可能需要更复杂的内存管理策略：
- 本地大块内存注册: 每个
MemoryPoolServer启动时，一次性分配并注册一大块物理连续的内存（例如，使用mmap与MAP_ANONYMOUS | MAP_PRIVATE | MAP_HUGETLB来尝试分配大页，并确保它被锁定）。 - 池内细粒度管理: 在这块已注册的大内存区域内部，可以实现一个自定义的内存分配器，例如：
- Free List: 管理空闲内存块的链表。
- Buddy System: 适合分配不同大小的内存块。
- Slab Allocator: 如果内存池主要用于存储固定大小的对象。
- 远程分配/释放: 客户端可以请求远程节点分配或释放内存。这需要通过RDMA Atomic操作来协调。例如，客户端可以通过RDMA CAS操作原子性地修改远程节点内存池的空闲列表指针或位图，实现无锁或低锁的远程内存分配。
C++代码骨架：RdmaMemoryPool
我们将创建一个RdmaMemoryPool类，它既可以作为Server端注册本地内存并共享信息，也可以作为Client端连接远程节点并执行RDMA操作。
// -------------------------------------------------------------------------
// 辅助结构体：RDMA CM ID 和连接信息
// -------------------------------------------------------------------------
struct RdmaConnectionInfo {
ibv_qp_init_attr qp_init_attr; // QP初始化属性
ibv_qp_attr qp_attr; // QP当前属性
ibv_qp* qp = nullptr; // 队列对
ibv_port_attr port_attr; // 端口属性
uint16_t lid; // 本地ID (InfiniBand)
union ibv_gid gid; // 全局ID (RoCE/InfiniBand)
uint32_t qp_num; // QP号
RdmaMemoryRegion* local_mr = nullptr; // 本地注册的内存区域
// 用于与远程节点交换的MR信息
struct RemoteMRInfo {
uint64_t addr;
uint32_t rkey;
size_t length;
};
std::vector<RemoteMRInfo> remote_mrs; // 存储远程节点的MR信息
};
// -------------------------------------------------------------------------
// RdmaMemoryPool: 核心内存池管理类
// -------------------------------------------------------------------------
class RdmaMemoryPool {
public:
RdmaMemoryPool(const std::string& dev_name = "") : context_(dev_name) {
// 分配并注册一大块内存作为本地内存池
// 实际应用中，这里可能使用mmap分配大页内存
pool_buffer_ = std::make_unique<char[]>(POOL_SIZE);
local_mr_ = std::make_unique<RdmaMemoryRegion>(context_, pool_buffer_.get(), POOL_SIZE);
std::cout << "Local memory pool allocated and registered. Size: " << POOL_SIZE << " bytes." << std::endl;
}
~RdmaMemoryPool() {
// 清理所有连接
for (auto& pair : connections_) {
disconnect_qp(pair.second.qp);
// rdma_destroy_id(pair.second.cm_id); // 如果使用rdma_cm_id管理
}
std::cout << "RDMA Memory Pool destroyed." << std::endl;
}
// ----------------------------------------
// Server端功能
// ----------------------------------------
void start_server(const std::string& server_address, const std::string& port) {
// 使用rdma_cm_id作为辅助连接通道
struct rdma_cm_id *listen_id = nullptr;
struct rdma_addrinfo hints, *res;
memset(&hints, 0, sizeof(hints));
hints.ai_flags = RAI_PASSIVE; // 服务器模式
hints.ai_port_space = RDMA_PS_TCP; // 基于TCP的RDMA CM
int ret = rdma_getaddrinfo(server_address.c_str(), port.c_str(), &hints, &res);
check_rdma_error(ret, "rdma_getaddrinfo for server failed");
ret = rdma_create_id(nullptr, &listen_id, nullptr, RDMA_PS_TCP);
check_rdma_error(ret, "rdma_create_id for server failed");
ret = rdma_bind_addr(listen_id, res->ai_src_addr);
check_rdma_error(ret, "rdma_bind_addr failed");
ret = rdma_listen(listen_id, 10); // 监听，最大连接数10
check_rdma_error(ret, "rdma_listen failed");
std::cout << "RDMA Server listening on " << server_address << ":" << port << std::endl;
while (true) {
struct rdma_cm_id *conn_id = nullptr;
ret = rdma_get_request(listen_id, &conn_id); // 等待客户端连接
if (ret) {
std::cerr << "rdma_get_request failed: " << strerror(errno) << std::endl;
continue;
}
std::cout << "Client connected." << std::endl;
// 创建QP
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; // 所有WR都生成WC
qp_init_attr.send_cq = context_.get_cq();
qp_init_attr.recv_cq = context_.get_cq();
qp_init_attr.cap.max_send_wr = MAX_WR;
qp_init_attr.cap.max_recv_wr = MAX_WR; // 虽然单边操作不用RQ，但QP要求
qp_init_attr.cap.max_send_sge = MAX_SGE;
qp_init_attr.cap.max_recv_sge = MAX_SGE;
ret = rdma_create_qp(conn_id, context_.get_pd(), &qp_init_attr);
check_rdma_error(ret, "rdma_create_qp for client failed");
// 交换MR信息
RdmaConnectionInfo::RemoteMRInfo local_mr_info = {
(uint64_t)local_mr_->get_addr(),
local_mr_->get_mr()->rkey,
local_mr_->get_length()
};
struct rdma_conn_param conn_param;
memset(&conn_param, 0, sizeof(conn_param));
conn_param.initiator_depth = 1; // 允许一个未完成的RDMA操作
conn_param.responder_resources = 1; // 允许一个远程节点响应
conn_id->qp->qp_num; // 获取QP号
// 在这里，我们可以通过rdma_cm_event (RDMA_CM_EVENT_CONNECT_REQUEST, RDMA_CM_EVENT_ESTABLISHED)
// 携带私有数据 (private_data) 来交换MR信息。
// 这是一个简化的示例，假设通过某种方式（例如，再次通过一个TCP socket）交换。
// 实际中，`rdma_conn_param.private_data` 和 `private_data_len` 可以用来传递这些信息。
// 这里为了简洁，假设后续通过一个TCP辅助通道完成。
// 接受连接
ret = rdma_accept(conn_id, &conn_param);
check_rdma_error(ret, "rdma_accept failed");
connections_[conn_id->qp->qp_num] = RdmaConnectionInfo();
connections_[conn_id->qp->qp_num].qp = conn_id->qp;
connections_[conn_id->qp->qp_num].local_mr = local_mr_.get();
// ... 存储其他信息，如远程MR信息（稍后接收）
}
rdma_freeaddrinfo(res);
rdma_destroy_id(listen_id);
}
// ----------------------------------------
// Client端功能
// ----------------------------------------
void connect_to_server(uint32_t node_id, const std::string& server_address, const std::string& port) {
struct rdma_cm_id *cm_id = nullptr;
struct rdma_addrinfo hints, *res;
memset(&hints, 0, sizeof(hints));
hints.ai_port_space = RDMA_PS_TCP; // 基于TCP的RDMA CM
int ret = rdma_getaddrinfo(server_address.c_str(), port.c_str(), &hints, &res);
check_rdma_error(ret, "rdma_getaddrinfo for client failed");
ret = rdma_create_id(nullptr, &cm_id, nullptr, RDMA_PS_TCP);
check_rdma_error(ret, "rdma_create_id for client failed");
// 创建QP
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 = context_.get_cq();
qp_init_attr.recv_cq = context_.get_cq();
qp_init_attr.cap.max_send_wr = MAX_WR;
qp_init_attr.cap.max_recv_wr = MAX_WR;
qp_init_attr.cap.max_send_sge = MAX_SGE;
qp_init_attr.cap.max_recv_sge = MAX_SGE;
ret = rdma_create_qp(cm_id, context_.get_pd(), &qp_init_attr);
check_rdma_error(ret, "rdma_create_qp for client failed");
// 交换MR信息 (这里同样简化，假设通过私有数据传递或辅助TCP通道)
// struct rdma_conn_param conn_param = {};
// conn_param.private_data = &local_mr_info;
// conn_param.private_data_len = sizeof(local_mr_info);
// ...
ret = rdma_connect(cm_id, nullptr); // 连接服务器
check_rdma_error(ret, "rdma_connect failed");
std::cout << "Client connected to " << server_address << ":" << port << std::endl;
connections_[node_id] = RdmaConnectionInfo();
connections_[node_id].qp = cm_id->qp;
connections_[node_id].local_mr = local_mr_.get();
// 这里需要接收远程MR信息并存储到 connections_[node_id].remote_mrs
// 实际应用中，可以通过rdma_cm_event获取连接事件，并处理私有数据。
// 或者在连接建立后，通过一个独立的TCP socket交换这些信息。
// 假设通过辅助TCP通道获取到了 remote_mr_info
RdmaConnectionInfo::RemoteMRInfo remote_server_mr_info = {
(uint64_t)0x12345678, // 假设的远程地址
0xABCD, // 假设的远程RKey
POOL_SIZE // 假设的远程长度
};
connections_[node_id].remote_mrs.push_back(remote_server_mr_info);
rdma_freeaddrinfo(res);
}
// ----------------------------------------
// RDMA单边读写操作
// ----------------------------------------
// RDMA Write: 将本地数据写入远程内存
void rdma_write(uint32_t node_id, size_t local_offset, size_t remote_offset, size_t length) {
if (connections_.find(node_id) == connections_.end() || connections_[node_id].remote_mrs.empty()) {
throw std::runtime_error("No connection or remote MR info for node " + std::to_string(node_id));
}
ibv_qp* qp = connections_[node_id].qp;
RdmaConnectionInfo::RemoteMRInfo& remote_mr_info = connections_[node_id].remote_mrs[0]; // 假设只有一个远程MR
ibv_sge sge;
sge.addr = (uint64_t)local_mr_->get_addr() + local_offset;
sge.length = length;
sge.lkey = local_mr_->get_mr()->lkey;
ibv_send_wr wr;
memset(&wr, 0, sizeof(wr));
wr.wr_id = 1; // 用户自定义ID，用于WC匹配
wr.sg_list = &sge;
wr.num_sge = 1;
wr.opcode = IBV_WR_RDMA_WRITE;
wr.send_flags = IBV_SEND_SIGNALED; // 要求生成WC
wr.wr.rdma.remote_addr = remote_mr_info.addr + remote_offset;
wr.wr.rdma.rkey = remote_mr_info.rkey;
ibv_send_wr* bad_wr = nullptr;
check_rdma_error(ibv_post_send(qp, &wr, &bad_wr), "Failed to post RDMA WRITE WR");
wait_for_completion(); // 等待WC
}
// RDMA Read: 从远程内存读取数据到本地
void rdma_read(uint32_t node_id, size_t local_offset, size_t remote_offset, size_t length) {
if (connections_.find(node_id) == connections_.end() || connections_[node_id].remote_mrs.empty()) {
throw std::runtime_error("No connection or remote MR info for node " + std::to_string(node_id));
}
ibv_qp* qp = connections_[node_id].qp;
RdmaConnectionInfo::RemoteMRInfo& remote_mr_info = connections_[node_id].remote_mrs[0];
ibv_sge sge;
sge.addr = (uint64_t)local_mr_->get_addr() + local_offset;
sge.length = length;
sge.lkey = local_mr_->get_mr()->lkey;
ibv_send_wr wr;
memset(&wr, 0, sizeof(wr));
wr.wr_id = 2; // 用户自定义ID
wr.sg_list = &sge;
wr.num_sge = 1;
wr.opcode = IBV_WR_RDMA_READ;
wr.send_flags = IBV_SEND_SIGNALED;
wr.wr.rdma.remote_addr = remote_mr_info.addr + remote_offset;
wr.wr.rdma.rkey = remote_mr_info.rkey;
ibv_send_wr* bad_wr = nullptr;
check_rdma_error(ibv_post_send(qp, &wr, &bad_wr), "Failed to post RDMA READ WR");
wait_for_completion();
}
// RDMA Atomic Compare-and-Swap
// local_val_ptr: 指向本地内存中存放比较值和新值的64位数据
// compare_val: 要比较的旧值
// swap_val: 如果比较成功，写入的新值
void rdma_atomic_cas(uint32_t node_id, size_t remote_offset, uint64_t compare_val, uint64_t swap_val) {
if (connections_.find(node_id) == connections_.end() || connections_[node_id].remote_mrs.empty()) {
throw std::runtime_error("No connection or remote MR info for node " + std::to_string(node_id));
}
ibv_qp* qp = connections_[node_id].qp;
RdmaConnectionInfo::RemoteMRInfo& remote_mr_info = connections_[node_id].remote_mrs[0];
// 原子操作不需要本地SGE，但需要一个虚拟SGE来满足API要求
ibv_sge sge;
sge.addr = 0; // 不使用本地缓冲区进行数据传输
sge.length = 0;
sge.lkey = local_mr_->get_mr()->lkey; // 仍然需要一个lkey
ibv_send_wr wr;
memset(&wr, 0, sizeof(wr));
wr.wr_id = 3; // 用户自定义ID
wr.sg_list = &sge;
wr.num_sge = 1;
wr.opcode = IBV_WR_ATOMIC_CMP_AND_SWP;
wr.send_flags = IBV_SEND_SIGNALED;
wr.wr.atomic.remote_addr = remote_mr_info.addr + remote_offset;
wr.wr.atomic.rkey = remote_mr_info.rkey;
wr.wr.atomic.compare_add = compare_val; // For CAS, this is the compare value
wr.wr.atomic.swap = swap_val; // For CAS, this is the swap value
ibv_send_wr* bad_wr = nullptr;
check_rdma_error(ibv_post_send(qp, &wr, &bad_wr), "Failed to post RDMA ATOMIC CAS WR");
wait_for_completion();
}
// 获取本地内存池的地址
char* get_local_pool_buffer() {
return pool_buffer_.get();
}
private:
RdmaContext context_;
std::unique_ptr<char[]> pool_buffer_;
std::unique_ptr<RdmaMemoryRegion> local_mr_;
std::map<uint32_t, RdmaConnectionInfo> connections_; // key: node_id
static const size_t POOL_SIZE = 16 * 1024 * 1024; // 16MB 内存池
static const int MAX_WR = 128; // Max Work Requests
static const int MAX_SGE = 1; // Max Scatter/Gather Entries per WR
// 等待一个工作完成事件
void wait_for_completion() {
ibv_wc wc;
int ret = 0;
do {
ret = ibv_poll_cq(context_.get_cq(), 1, &wc);
if (ret < 0) {
throw std::runtime_error("Failed to poll CQ: " + std::string(strerror(errno)));
}
} while (ret == 0); // 持续轮询直到有WC
if (wc.status != IBV_WC_SUCCESS) {
std::cerr << "WC status: " << ibv_wc_status_str(wc.status) << ", opcode: " << wc.opcode << std::endl;
throw std::runtime_error("RDMA operation failed with status: " + std::string(ibv_wc_status_str(wc.status)));
}
std::cout << "RDMA operation (WR ID: " << wc.wr_id << ") completed successfully." << std::endl;
}
void disconnect_qp(ibv_qp* qp) {
if (!qp) return;
struct ibv_qp_attr qp_attr;
memset(&qp_attr, 0, sizeof(qp_attr));
qp_attr.qp_state = IBV_QPS_SQD; // 将QP置于停止发送状态
check_rdma_error(ibv_modify_qp(qp, &qp_attr, IBV_QP_STATE), "Failed to modify QP to SQD");
qp_attr.qp_state = IBV_QPS_ERR; // 将QP置于错误状态，强制断开
check_rdma_error(ibv_modify_qp(qp, &qp_attr, IBV_QP_STATE), "Failed to modify QP to ERR");
check_rdma_error(ibv_destroy_qp(qp), "Failed to destroy QP");
}
};
// -------------------------------------------------------------------------
// 简单示例：Server 和 Client 的主函数
// -------------------------------------------------------------------------
int main(int argc, char* argv[]) {
if (argc < 2) {
std::cerr << "Usage: " << argv[0] << " <server|client> [server_ip] [server_port]" << std::endl;
return 1;
}
std::string mode = argv[1];
try {
if (mode == "server") {
RdmaMemoryPool server_pool;
// Server启动后，需要某种机制告诉客户端其RKey等信息
// 假设我们通过一个辅助TCP通道或者手工配置来完成
std::string server_ip = (argc > 2) ? argv[2] : "0.0.0.0";
std::string server_port = (argc > 3) ? argv[3] : "18888";
server_pool.start_server(server_ip, server_port); // 这会阻塞等待连接
} else if (mode == "client") {
if (argc < 4) {
std::cerr << "Usage for client: " << argv[0] << " client <server_ip> <server_port>" << std::endl;
return 1;
}
std::string server_ip = argv[2];
std::string server_port = argv[3];
RdmaMemoryPool client_pool;
client_pool.connect_to_server(1, server_ip, server_port); // 节点ID为1
// 示例：Client写入数据到Server的内存
const char* test_data = "Hello from Client RDMA Write!";
size_t data_len = strlen(test_data) + 1;
size_t remote_offset = 0; // 写入Server内存池的起始位置
// 先将数据拷贝到本地内存池
memcpy(client_pool.get_local_pool_buffer(), test_data, data_len);
std::cout << "Client: Writing '" << test_data << "' to remote offset " << remote_offset << std::endl;
client_pool.rdma_write(1, 0, remote_offset, data_len); // 本地偏移0，写入远程偏移0
// 示例：Client从Server内存读取数据
char read_buf[256];
memset(read_buf, 0, sizeof(read_buf));
std::cout << "Client: Reading from remote offset " << remote_offset << " with length " << data_len << std::endl;
client_pool.rdma_read(1, 0, remote_offset, data_len); // 读取到本地偏移0
memcpy(read_buf, client_pool.get_local_pool_buffer(), data_len);
std::cout << "Client: Read data from remote: '" << read_buf << "'" << std::endl;
// 示例：原子操作 (假设远程偏移 1024 处有一个64位计数器)
uint64_t counter_offset = 1024;
uint64_t expected_val = 0; // 假设期望当前值为0
uint64_t new_val = 1; // 如果是0，更新为1
std::cout << "Client: Attempting Atomic CAS on remote offset " << counter_offset
<< " (Compare: " << expected_val << ", Swap: " << new_val << ")" << std::endl;
// CAS操作返回的是远程内存的原始值。我们需要一个地方来存储这个值。
// RDMA Atomic操作的WR本身不包含SGE用于返回旧值，但WC中可以包含特定信息
// 实际应用中，原子操作后的值通常需要通过一个独立的RDMA READ来获取。
// 为了简洁，这里仅演示CAS的提交。
client_pool.rdma_atomic_cas(1, counter_offset, expected_val, new_val);
} else {
std::cerr << "Invalid mode. Use 'server' or 'client'." << std::endl;
return 1;
}
} catch (const std::exception& e) {
std::cerr << "Error: " << e.what() << std::endl;
return 1;
}
return 0;
}注意：
- 上述代码简化了RDMA CM的事件处理和私有数据交换部分。在实际生产环境中，需要实现一个完整的事件循环来处理
rdma_cm_event，并利用rdma_conn_param.private_data来交换MR信息和QP号等关键元数据。 - 内存池的分配/释放逻辑被简化为一次性分配大块内存。真正的分布式内存池需要一个复杂的内存管理器来处理碎片、并发分配和释放。
wait_for_completion()采用阻塞轮询。在高性能应用中，通常会使用非阻塞轮询或事件通知机制（如ibv_get_cq_event）来提高效率。
零拷贝读写与性能优势
传统TCP/IP的拷贝开销分析
在传统的TCP/IP通信中，数据从应用程序发送到网络通常涉及以下步骤：
- 用户态到内核态拷贝：应用程序将数据从用户缓冲区拷贝到内核的socket缓冲区。
- 内核态到网卡拷贝：操作系统将数据从内核socket缓冲区拷贝到网卡驱动的缓冲区。
- 网卡发送：网卡将数据发送到网络。
在接收端，也存在类似的反向拷贝过程。这些拷贝操作不仅消耗CPU周期，还增加了数据在内存中的停留时间，从而提高了端到端延迟。
RDMA如何消除拷贝：用户空间直接访问、DMA引擎
RDMA通过以下机制彻底消除了这些拷贝：
- 内存注册与页锁定：应用程序将内存区域注册给HCA，HCA获得对这些内存的直接访问权限，并防止操作系统将其换出。
- 用户空间直接访问：应用程序可以直接在用户空间构建WR，并将其提交给HCA，无需经过内核。
- DMA引擎：HCA内置的DMA引擎负责在注册内存与网络之间直接传输数据，完全绕过了CPU和操作系统内核。
这种机制带来的性能优势是显著的：
- 极低延迟：RDMA通信的端到端延迟通常在微秒甚至亚微秒级别，远低于传统TCP/IP的几十到几百微秒。
- 高带宽：RDMA能够充分利用网络硬件的带宽，达到100Gbps甚至更高。
- CPU利用率降低：由于CPU无需参与数据传输，其资源可以完全用于计算，提高了系统整体的计算效率。
延迟与带宽提升：微秒级延迟，Gbps/Tbps带宽
| 特性 | 传统TCP/IP (10GbE) | RDMA (RoCE/InfiniBand 100GbE) |
|---|---|---|
| 延迟 | 几十到几百微秒 | 1-5微秒 |
| 带宽 | 10 Gbps | 100 Gbps 及更高 |
| CPU | 高 CPU 占用 | 低 CPU 占用 (卸载) |
| 内存 | 多次拷贝 | 零拷贝 |
挑战与最佳实践
尽管RDMA带来了巨大的性能提升，但在实际应用中也面临一些挑战，需要遵循最佳实践：
- 内存对齐与页锁定：
ibv_reg_mr要求注册的内存是页对齐的，并且会被操作系统锁定在物理内存中。这意味着这部分内存不能被交换到磁盘，可能会影响系统整体的内存管理。合理规划内存使用，避免过度注册。 - 错误处理与恢复：RDMA操作是异步的，错误可能在提交WR之后一段时间才通过WC报告。需要健全的错误处理机制来检查WC状态，并对失败的操作进行重试或恢复。网络分区、HCA故障等硬件错误也需要考虑。
- 连接管理与扩展性：在大型分布式系统中，管理大量的QP和连接可能变得复杂。RDMA CM简化了连接建立，但对于大规模集群，仍需精心设计连接池和断开重连逻辑。
- 并发控制：虽然RDMA单边操作绕过了远程CPU，但如果多个发起方同时写入同一个远程内存区域，仍然需要同步机制来保证数据一致性。RDMA Atomic操作是实现分布式锁、计数器等并发原语的强大工具，但它们通常只支持64位原子操作，且性能不如批量读写。对于更复杂的数据结构，可能需要结合RDMA Atomic和两阶段提交等协议。
- 安全性：RDMA允许直接内存访问，这意味着如果配置不当，恶意程序可能绕过操作系统安全机制直接访问甚至修改远程内存。因此，RDMA网络的隔离、验证和权限管理至关重要。
- 资源限制：HCA上的QP、MR、CQ等资源数量是有限的。应用程序需要合理规划和复用这些资源，避免耗尽。
应用场景与未来展望
RDMA技术及其带来的零拷贝、低延迟、高吞吐量优势，使其在多个高性能领域得到广泛应用：
- 分布式数据库与键值存储：如Redis、Memcached、Cassandra等，利用RDMA加速数据访问和复制，实现超低延迟的读写操作。
- 高性能计算 (HPC) 与机器学习：在科学计算、仿真、深度学习训练中，节点间大量数据的交换是性能瓶颈。RDMA能够显著加速MPI等通信库，提升大规模并行计算效率。
- 内存计算与数据分析：如Apache Spark、Apache Flink等内存计算框架，通过RDMA加速Shuffle操作和数据共享，提升实时数据处理能力。
- 存储系统：NVMe over Fabrics (NVMe-oF) 利用RDMA将NVMe SSD的低延迟特性扩展到网络，实现高性能的共享存储。
- 金融交易系统：对延迟要求极高的场景，RDMA可以提供毫秒甚至微秒级的交易响应速度。
随着RoCE (RDMA over Converged Ethernet) 技术的普及，RDMA不再局限于昂贵的InfiniBand网络，而是可以在标准以太网上实现，这极大地降低了RDMA的部署门槛，使其在数据中心和云计算环境中得到更广泛的应用。未来，RDMA有望与新型内存技术（如持久内存）结合，构建更高效、更强大的分布式内存计算平台。
RDMA通过其独特的零拷贝和CPU卸载机制，为C++开发者提供了构建高性能分布式系统的强大工具。掌握RDMA单边操作的精髓，并将其应用于跨节点内存池的实现，是迈向下一代高性能计算架构的关键一步。