MinIO源码深度解读-第11章第2节-RDMA-传输层实现-RDMA-连接管理与数据传输完整源码

MinIO源码深度解读 – 第11章第2节：RDMA 传输层实现——RDMA 连接管理与数据传输完整源码分析

本文深入分析 rdma-transport.go 中 RDMA 传输层的完整实现，包括连接管理、数据传输和错误处理。

本文导航

• 连接管理架构
• 连接创建流程
• 数据传输实现
• 内存注册与缓存
• 错误处理与恢复
• 性能监控

连接管理架构

连接管理核心接口

// cmd/rdma-transport.go

// RDMA 连接接口
type RDMAConnection interface {
// 连接状态
IsConnected() bool
GetState() ConnectionState

// 数据传输
Send(data []byte) error
Recv(buffer []byte) (int, error)
Write(key uint64, offset int64, size int64) error
Read(key uint64, buffer []byte, size int64) error

// 异步操作
AsyncSend(data []byte) (*SendRequest, error)
AsyncRecv(buffer []byte) (*RecvRequest, error)

// 完成等待
Wait() <-chan struct{}

// 关闭连接
Close() error
}

// 连接状态
type ConnectionState int

const (
ConnectionStateIdle ConnectionState = iota
ConnectionStateConnecting
ConnectionStateActive
ConnectionStateClosing
ConnectionStateError
)

// RDMA 传输层实现
type RDMATransport struct {
// 设备上下文
ctx *ibv_context

// 端口信息
port int

// QP 表
qpTable map[string]QPTable

// 连接表
connections map[string]*RDMAConnection

// 内存池
memoryPool *MemoryPool

// CQ (Completion Queue)
cq ibv_cq

// 统计信息
stats RDMAStats

// 锁
mu sync.RWMutex

// 停止通道
stopCh chan struct{}
}

// QP 表（Queue Pair 表）
type QPTable struct {
// 本地 QP
local *ibv_qp

// 远程 QP
remote *ibv_qp

// 发送队列
sendQ chan *SendRequest

// 接收队列
recvQ chan *RecvRequest

// 完成队列
cq ibv_cq
}

连接管理器

// cmd/rdma-transport.go

// 连接管理器
type ConnectionManager struct {
// 活跃连接池
activeConnections *ConnectionPool

// 空闲连接池
idleConnections *ConnectionPool

// 最大连接数
maxConnections int

// 连接超时
timeout time.Duration

// 锁
mu sync.RWMutex

// 统计
stats ConnectionStats
}

// 创建连接
func (cm *ConnectionManager) CreateConnection(endpoint string) (*RDMAConnection, error) {
// 1. 检查最大连接数
if cm.activeConnections.Size() >= cm.maxConnections {
return nil, errMaxConnectionsExceeded
}

// 2. 解析端点
host, port, err := parseEndpoint(endpoint)
if err != nil {
return nil, err
}

// 3. 创建 RDMA 设备上下文
ctx, err := createRDMAContext()
if err != nil {
return nil, err
}

// 4. 查找目标 IP
addr, err := net.ResolveIPAddr("ipv4", host)
if err != nil {
return nil, err
}

// 5. 创建 QP (Queue Pair)
qp, err := createQueuePair(ctx, 0)
if err != nil {
return nil, err
}

// 6. 初始化 QP 状态
if err := initializeQP(qp, addr, port); err != nil {
return nil, err
}

// 7. 创建连接对象
conn := &RDMAConnection{
ctx: ctx,
qp: qp,
addr: addr,
port: port,
state: ConnectionStateConnecting,
sendQ: make(chan *SendRequest, 128),
recvQ: make(chan *RecvRequest, 128),
done: make(chan struct{}),
}

// 8. 添加到活跃连接池
cm.mu.Lock()
cm.activeConnections.Add(endpoint, conn)
cm.stats.TotalConnections++
cm.mu.Unlock()

// 9. 启动异步收发
go cm.asyncSendLoop(conn)
go cm.asyncRecvLoop(conn)

return conn, nil
}

// 获取连接
func (cm *ConnectionManager) GetConnection(endpoint string) (*RDMAConnection, bool) {
cm.mu.RLock()
defer cm.mu.RUnlock()

// 1. 从活跃连接查找
if conn, ok := cm.activeConnections.Get(endpoint); ok {
if conn.IsConnected() {
return conn, true
}
}

// 2. 从空闲连接查找
if conn, ok := cm.idleConnections.Get(endpoint); ok {
if conn.IsConnected() {
cm.activeConnections.Add(endpoint, conn)
cm.idleConnections.Remove(endpoint)
return conn, true
}
}

// 3. 连接不存在或已断开
return nil, false
}

// 关闭连接
func (cm *ConnectionManager) CloseConnection(endpoint string) error {
cm.mu.Lock()
defer cm.mu.Unlock()

conn, ok := cm.activeConnections.Get(endpoint)
if !ok {
return errConnectionNotFound
}

// 1. 停止异步循环
close(conn.done)

// 2. 关闭 QP
if err := conn.qp.Close(); err != nil {
return err
}

// 3. 移动到空闲池或销毁
conn.state = ConnectionStateIdle
cm.idleConnections.Add(endpoint, conn)
cm.stats.TotalConnections—

return nil
}

连接创建流程

完整连接创建流程

// cmd/rdma-transport.go

// 完整连接创建流程
func (t *RDMATransport) connectToEndpoint(endpoint string) (*RDMAConnection, error) {
// 1. 解析端点
host, port, err := parseEndpoint(endpoint)
if err != nil {
return nil, err
}

// 2. 创建 RDMA 设备上下文
dev, err := t.findRDAMDevice(host)
if err != nil {
return nil, err
}

// 3. 打开设备
ctx, err := dev.Open()
if err != nil {
return nil, err
}
t.ctx = ctx

// 4. 获取端口信息
portAttr, err := ctx.GetPortAttr(t.port)
if err != nil {
return nil, err
}

// 5. 创建 Protection Domain
pd, err := ctx.CreateProtectionDomain()
if err != nil {
return nil, err
}

// 6. 创建 Completion Queue
cq, err := ctx.CreateCompletionQueue(128)
if err != nil {
return nil, err
}
t.cq = cq

// 7. 创建 Queue Pair (QP)
qp, err := pd.CreateQueuePair(ibvQPT_RC, 128, 128, cq, cq)
if err != nil {
return nil, err
}

// 8. 查询 GID (Global Identifier)
gid, err := ctx.GetGID(t.port, 0)
if err != nil {
return nil, err
}

// 9. 创建 RD address handle
rkey, err := pd.CreateRKey(qp, ibvAccess_LOCAL_WRITE|
ibvAccess_REMOTE_WRITE|ibvAccess_REMOTE_READ)
if err != nil {
return nil, err
}

// 10. 初始化 QP 状态
qpAttr := &ibv_qp_attr{
AttrMask: ibvQpAttrPathMtu | ibvQpAttrRoute | ibvQpAttrQkey,
PathMtu: portAttr.ActiveMtu,
Qkey: 0x00000001,
Route: &ibv_route{
Gid: gid,
Port: t.port,
DLid: portAttr.Lid,
SL: 0,
TrafficClass: 0,
HopLimit: 64,
},
}

if err := qp.ModifyState(qpAttr); err != nil {
return nil, err
}

// 11. 转换 QP 状态到 RTR (Ready to Receive)
rtrAttr := &ibv_qp_attr{
State: ibvStateRTR,
PathMtu: portAttr.ActiveMtu,
}
if err := qp.SetAttr(rtrAttr); err != nil {
return nil, err
}

// 12. 转换 QP 状态到 RTS (Ready to Send)
rtsAttr := &ibv_qp_attr{
State: ibvStateRTS,
Timeout: 14,
RetryCnt: 7,
RnrRetry: 7,
SqPsn: 0,
}
if err := qp.SetAttr(rtsAttr); err != nil {
return nil, err
}

// 13. 创建连接对象
conn := &RDMAConnection{
ctx: ctx,
qp: qp,
rkey: rkey,
state: ConnectionStateActive,
sendQ: make(chan *SendRequest, 128),
recvQ: make(chan *RecvRequest, 128),
done: make(chan struct{}),
}

// 14. 添加到连接管理器
t.connections[endpoint] = conn

// 15. 启动异步处理
go t.processSendQueue(conn)
go t.processRecvQueue(conn)

return conn, nil
}

QP 状态转换流程

#mermaid-svg-8sb1mjUDJidfnXwY{font-family:\”trebuchet ms\”,verdana,arial,sans-serif;font-size:16px;fill:#333;}@keyframes edge-animation-frame{from{stroke-dashoffset:0;}}@keyframes dash{to{stroke-dashoffset:0;}}#mermaid-svg-8sb1mjUDJidfnXwY .edge-animation-slow{stroke-dasharray:9,5!important;stroke-dashoffset:900;animation:dash 50s linear infinite;stroke-linecap:round;}#mermaid-svg-8sb1mjUDJidfnXwY .edge-animation-fast{stroke-dasharray:9,5!important;stroke-dashoffset:900;animation:dash 20s linear infinite;stroke-linecap:round;}#mermaid-svg-8sb1mjUDJidfnXwY .error-icon{fill:#552222;}#mermaid-svg-8sb1mjUDJidfnXwY .error-text{fill:#552222;stroke:#552222;}#mermaid-svg-8sb1mjUDJidfnXwY .edge-thickness-normal{stroke-width:1px;}#mermaid-svg-8sb1mjUDJidfnXwY .edge-thickness-thick{stroke-width:3.5px;}#mermaid-svg-8sb1mjUDJidfnXwY .edge-pattern-solid{stroke-dasharray:0;}#mermaid-svg-8sb1mjUDJidfnXwY .edge-thickness-invisible{stroke-width:0;fill:none;}#mermaid-svg-8sb1mjUDJidfnXwY .edge-pattern-dashed{stroke-dasharray:3;}#mermaid-svg-8sb1mjUDJidfnXwY .edge-pattern-dotted{stroke-dasharray:2;}#mermaid-svg-8sb1mjUDJidfnXwY .marker{fill:#333333;stroke:#333333;}#mermaid-svg-8sb1mjUDJidfnXwY .marker.cross{stroke:#333333;}#mermaid-svg-8sb1mjUDJidfnXwY svg{font-family:\”trebuchet ms\”,verdana,arial,sans-serif;font-size:16px;}#mermaid-svg-8sb1mjUDJidfnXwY p{margin:0;}#mermaid-svg-8sb1mjUDJidfnXwY defs #statediagram-barbEnd{fill:#333333;stroke:#333333;}#mermaid-svg-8sb1mjUDJidfnXwY g.stateGroup text{fill:#9370DB;stroke:none;font-size:10px;}#mermaid-svg-8sb1mjUDJidfnXwY g.stateGroup text{fill:#333;stroke:none;font-size:10px;}#mermaid-svg-8sb1mjUDJidfnXwY g.stateGroup .state-title{font-weight:bolder;fill:#131300;}#mermaid-svg-8sb1mjUDJidfnXwY g.stateGroup rect{fill:#ECECFF;stroke:#9370DB;}#mermaid-svg-8sb1mjUDJidfnXwY g.stateGroup line{stroke:#333333;stroke-width:1;}#mermaid-svg-8sb1mjUDJidfnXwY .transition{stroke:#333333;stroke-width:1;fill:none;}#mermaid-svg-8sb1mjUDJidfnXwY .stateGroup .composit{fill:white;border-bottom:1px;}#mermaid-svg-8sb1mjUDJidfnXwY .stateGroup .alt-composit{fill:#e0e0e0;border-bottom:1px;}#mermaid-svg-8sb1mjUDJidfnXwY .state-note{stroke:#aaaa33;fill:#fff5ad;}#mermaid-svg-8sb1mjUDJidfnXwY .state-note text{fill:black;stroke:none;font-size:10px;}#mermaid-svg-8sb1mjUDJidfnXwY .stateLabel .box{stroke:none;stroke-width:0;fill:#ECECFF;opacity:0.5;}#mermaid-svg-8sb1mjUDJidfnXwY .edgeLabel .label rect{fill:#ECECFF;opacity:0.5;}#mermaid-svg-8sb1mjUDJidfnXwY .edgeLabel{background-color:rgba(232,232,232, 0.8);text-align:center;}#mermaid-svg-8sb1mjUDJidfnXwY .edgeLabel p{background-color:rgba(232,232,232, 0.8);}#mermaid-svg-8sb1mjUDJidfnXwY .edgeLabel rect{opacity:0.5;background-color:rgba(232,232,232, 0.8);fill:rgba(232,232,232, 0.8);}#mermaid-svg-8sb1mjUDJidfnXwY .edgeLabel .label text{fill:#333;}#mermaid-svg-8sb1mjUDJidfnXwY .label div .edgeLabel{color:#333;}#mermaid-svg-8sb1mjUDJidfnXwY .stateLabel text{fill:#131300;font-size:10px;font-weight:bold;}#mermaid-svg-8sb1mjUDJidfnXwY .node circle.state-start{fill:#333333;stroke:#333333;}#mermaid-svg-8sb1mjUDJidfnXwY .node .fork-join{fill:#333333;stroke:#333333;}#mermaid-svg-8sb1mjUDJidfnXwY .node circle.state-end{fill:#9370DB;stroke:white;stroke-width:1.5;}#mermaid-svg-8sb1mjUDJidfnXwY .end-state-inner{fill:white;stroke-width:1.5;}#mermaid-svg-8sb1mjUDJidfnXwY .node rect{fill:#ECECFF;stroke:#9370DB;stroke-width:1px;}#mermaid-svg-8sb1mjUDJidfnXwY .node polygon{fill:#ECECFF;stroke:#9370DB;stroke-width:1px;}#mermaid-svg-8sb1mjUDJidfnXwY #statediagram-barbEnd{fill:#333333;}#mermaid-svg-8sb1mjUDJidfnXwY .statediagram-cluster rect{fill:#ECECFF;stroke:#9370DB;stroke-width:1px;}#mermaid-svg-8sb1mjUDJidfnXwY .cluster-label,#mermaid-svg-8sb1mjUDJidfnXwY .nodeLabel{color:#131300;}#mermaid-svg-8sb1mjUDJidfnXwY .statediagram-cluster rect.outer{rx:5px;ry:5px;}#mermaid-svg-8sb1mjUDJidfnXwY .statediagram-state .divider{stroke:#9370DB;}#mermaid-svg-8sb1mjUDJidfnXwY .statediagram-state .title-state{rx:5px;ry:5px;}#mermaid-svg-8sb1mjUDJidfnXwY .statediagram-cluster.statediagram-cluster .inner{fill:white;}#mermaid-svg-8sb1mjUDJidfnXwY .statediagram-cluster.statediagram-cluster-alt .inner{fill:#f0f0f0;}#mermaid-svg-8sb1mjUDJidfnXwY .statediagram-cluster .inner{rx:0;ry:0;}#mermaid-svg-8sb1mjUDJidfnXwY .statediagram-state rect.basic{rx:5px;ry:5px;}#mermaid-svg-8sb1mjUDJidfnXwY .statediagram-state rect.divider{stroke-dasharray:10,10;fill:#f0f0f0;}#mermaid-svg-8sb1mjUDJidfnXwY .note-edge{stroke-dasharray:5;}#mermaid-svg-8sb1mjUDJidfnXwY .statediagram-note rect{fill:#fff5ad;stroke:#aaaa33;stroke-width:1px;rx:0;ry:0;}#mermaid-svg-8sb1mjUDJidfnXwY .statediagram-note rect{fill:#fff5ad;stroke:#aaaa33;stroke-width:1px;rx:0;ry:0;}#mermaid-svg-8sb1mjUDJidfnXwY .statediagram-note text{fill:black;}#mermaid-svg-8sb1mjUDJidfnXwY .statediagram-note .nodeLabel{color:black;}#mermaid-svg-8sb1mjUDJidfnXwY .statediagram .edgeLabel{color:red;}#mermaid-svg-8sb1mjUDJidfnXwY #dependencyStart,#mermaid-svg-8sb1mjUDJidfnXwY #dependencyEnd{fill:#333333;stroke:#333333;stroke-width:1;}#mermaid-svg-8sb1mjUDJidfnXwY .statediagramTitleText{text-anchor:middle;font-size:18px;fill:#333;}#mermaid-svg-8sb1mjUDJidfnXwY :root{–mermaid-font-family:\”trebuchet ms\”,verdana,arial,sans-serif;}

创建 QP

转换到 RTR

转换到 RTS

错误

关闭连接

重置

StateINIT

StateRTR

StateRTS

StateERR

Ready to Receive接收数据准备

Ready to Send发送数据准备

数据传输实现

Send 操作

// cmd/rdma-transport.go

// Send 请求
type SendRequest struct {
// 数据缓冲区
data []byte

// 内存区域信息
rkey uint64
lkey uint64

// 完成回调
done chan struct{}

// 错误
err error
}

// Send 数据
func (conn *RDMAConnection) Send(data []byte) error {
// 1. 检查连接状态
if conn.state != ConnectionStateActive {
return errConnectionNotActive
}

// 2. 注册内存
mr, err := conn.ctx.RegisterMemory(data)
if err != nil {
return err
}
defer conn.ctx.DeregisterMemory(mr)

// 3. 创建 Send Request
req := &SendRequest{
data: data,
rkey: mr.Key,
lkey: mr.LKey,
done: make(chan struct{}),
}

// 4. 发送到发送队列
select {
case conn.sendQ <- req:
// 发送成功，等待完成
<-req.done
return req.err
case <-conn.done:
return errConnectionClosed
}
}

// 处理发送队列
func (t *RDMATransport) processSendQueue(conn *RDMAConnection) {
for req := range conn.sendQ {
// 1. 构建 Work Request
wr := &ibv_send_wr{
Op: ibvOpSend,
Srqn: 0,
DmaLkey: req.lkey,
DmaRkey: req.rkey,
RemoteAddr: uint64(req.remoteAddr),
DmaLen: uint32(len(req.data)),
CompFlags: ibvCompFlagSignaled,
Done: req.done,
Err: &req.err,
}

// 2. 发送到 WQ (Work Queue)
badWr, err := conn.qp.PostSend(wr)
if err != nil {
req.err = err
close(req.done)
continue
}

// 3. 如果有坏请求，处理错误
if badWr != nil {
req.err = fmt.Errorf("send failed, bad wr")
close(req.done)
}
}
}

// Recv 操作
type RecvRequest struct {
// 接收缓冲区
buffer []byte

// 内存区域信息
lkey uint64

// 完成回调
done chan struct{}

// 接收字节数
bytesRead int

// 错误
err error
}

// Recv 数据
func (conn *RDMAConnection) Recv(buffer []byte) (int, error) {
// 1. 检查连接状态
if conn.state != ConnectionStateActive {
return 0, errConnectionNotActive
}

// 2. 注册内存
mr, err := conn.ctx.RegisterMemory(buffer)
if err != nil {
return 0, err
}
defer conn.ctx.DeregisterMemory(mr)

// 3. 创建 Recv Request
req := &RecvRequest{
buffer: buffer,
lkey: mr.LKey,
done: make(chan struct{}),
}

// 4. 发送到接收队列
select {
case conn.recvQ <- req:
// 接收成功，等待完成
<-req.done
return req.bytesRead, req.err
case <-conn.done:
return 0, errConnectionClosed
}
}

// 处理接收队列
func (t *RDMATransport) processRecvQueue(conn *RDMAConnection) {
for req := range conn.recvQ {
// 1. 构建 Receive Work Request
wr := &ibv_recv_wr{
Op: ibvOpRecv,
Lkey: req.lkey,
DmaLen: uint32(len(req.buffer)),
Done: req.done,
Err: &req.err,
BytesRead: &req.bytesRead,
}

// 2. 发送到 RQ (Receive Queue)
badWr, err := conn.qp.PostRecv(wr)
if err != nil {
req.err = err
close(req.done)
continue
}

// 3. 如果有坏请求，处理错误
if badWr != nil {
req.err = fmt.Errorf("recv failed, bad wr")
close(req.done)
}
}
}

异步传输实现

// cmd/rdma-transport.go

// 异步发送
func (conn *RDMAConnection) AsyncSend(data []byte) (*SendRequest, error) {
// 1. 注册内存
mr, err := conn.ctx.RegisterMemory(data)
if err != nil {
return nil, err
}

// 2. 创建请求
req := &SendRequest{
data: data,
rkey: mr.Key,
lkey: mr.LKey,
done: make(chan struct{}),
}

// 3. 发送到发送队列（非阻塞）
select {
case conn.sendQ <- req:
return req, nil
case <-conn.done:
return nil, errConnectionClosed
}
}

// 异步接收
func (conn *RDMAConnection) AsyncRecv(buffer []byte) (*RecvRequest, error) {
// 1. 注册内存
mr, err := conn.ctx.RegisterMemory(buffer)
if err != nil {
return nil, err
}

// 2. 创建请求
req := &RecvRequest{
buffer: buffer,
lkey: mr.LKey,
done: make(chan struct{}),
}

// 3. 发送到接收队列（非阻塞）
select {
case conn.recvQ <- req:
return req, nil
case <-conn.done:
return nil, errConnectionClosed
}
}

// 等待异步完成
func (req *SendRequest) Wait() error {
<-req.done
return req.err
}

func (req *RecvRequest) Wait() (int, error) {
<-req.done
return req.bytesRead, req.err
}

内存注册与缓存

内存池实现

// cmd/rdma-memory.go

// 内存池
type MemoryPool struct {
// 内存块列表
blocks []*MemoryBlock

// 总大小
totalSize int

// 已用大小
usedSize int

// 锁
mu sync.Mutex
}

// 内存块
type MemoryBlock struct {
// 内存地址
addr uintptr

// 大小
size int

// 内存区域信息
region MemoryRegion

// 是否可用
available bool
}

// 内存区域
type MemoryRegion struct {
// LKey (本地键)
LKey uint64

// RKey (远程键)
RKey uint64

// 内存地址
Addr uintptr

// 大小
Size int64
}

// 创建内存池
func NewMemoryPool(size int) *MemoryPool {
// 1. 对齐到页大小
alignedSize := ((size + 4095) / 4096) * 4096

// 2. 分配内存
buf := make([]byte, alignedSize)

return &MemoryPool{
blocks: []*MemoryBlock{
{
addr: uintptr(unsafe.Pointer(&buf[0])),
size: alignedSize,
available: true,
},
},
totalSize: alignedSize,
}
}

// 注册内存
func (mp *MemoryPool) Register(addr uintptr, size int64) (MemoryRegion, error) {
mp.mu.Lock()
defer mp.mu.Unlock()

// 1. 查找可用内存块
block := mp.findBlock(addr, size)
if block == nil {
return MemoryRegion{}, errMemoryNotRegistered
}

// 2. 注册内存
region, err := block.region.Register(addr, size)
if err != nil {
return MemoryRegion{}, err
}

// 3. 标记为已用
block.available = false
mp.usedSize += int(size)

return region, nil
}

// 取消注册内存
func (mp *MemoryPool) Deregister(region MemoryRegion) error {
mp.mu.Lock()
defer mp.mu.Unlock()

// 1. 查找内存块
block := mp.findBlockByRegion(region)
if block == nil {
return errBlockNotFound
}

// 2. 取消注册
if err := block.region.Deregister(); err != nil {
return err
}

// 3. 标记为可用
block.available = true
mp.usedSize -= int(region.Size)

return nil
}

// 查找内存块
func (mp *MemoryPool) findBlock(addr uintptr, size int64) *MemoryBlock {
for _, block := range mp.blocks {
if block.available &&
block.addr <= addr &&
int64(block.addr)+int64(block.size) >= addr+size {
return block
}
}
return nil
}

// 查找内存块（按区域）
func (mp *MemoryPool) findBlockByRegion(region MemoryRegion) *MemoryBlock {
for _, block := range mp.blocks {
if block.region.Addr == region.Addr &&
block.region.LKey == region.LKey {
return block
}
}
return nil
}

错误处理与恢复

错误处理机制

// cmd/rdma-errors.go

// RDMA 错误类型
type RDMAError struct {
Code RDMAErrorCode
Message string
Cause error
}

// RDMA 错误码
type RDMAErrorCode int

const (
ErrRDMAUnknown RDMAErrorCode = iota
ErrRDMAConnectionFailed
ErrRDMAInvalidEndpoint
ErrRDMAOutOfMemory
ErrRDMAMemoryRegistrationFailed
ErrRDMAQPCreationFailed
ErrRDMAQueueFull
ErrRDMACompletionTimeout
ErrRDMAConnectionClosed
ErrRDMAInvalidState
)

// 创建 RDMA 错误
func NewRDMAError(code RDMAErrorCode, message string, cause error) *RDMAError {
return &RDMAError{
Code: code,
Message: message,
Cause: cause,
}
}

// 错误信息
func (e *RDMAError) Error() string {
if e.Cause != nil {
return fmt.Sprintf("RDMA error [%d]: %s – %v", e.Code, e.Message, e.Cause)
}
return fmt.Sprintf("RDMA error [%d]: %s", e.Code, e.Message)
}

// 连接恢复机制
type ConnectionRecovery struct {
// 最大重试次数
maxRetries int

// 重试间隔
retryInterval time.Duration

// 当前重试次数
retryCount int

// 上一次成功时间
lastSuccess time.Time
}

// 重试连接
func (cr *ConnectionRecovery) RetryConnect(endpoint string) error {
for i := 0; i < cr.maxRetries; i++ {
// 1. 尝试连接
conn, err := globalRDAManager.Connect(endpoint)
if err == nil {
// 2. 连接成功
cr.retryCount = 0
cr.lastSuccess = time.Now()
return nil
}

// 3. 等待并重试
time.Sleep(cr.retryInterval * time.Duration(i+1))
cr.retryCount++
}

return errMaxRetriesExceeded
}

// 检查连接健康状态
func (conn *RDMAConnection) HealthCheck() bool {
// 1. 检查连接状态
if conn.state != ConnectionStateActive {
return false
}

// 2. 检查最后活动时间
if time.Since(conn.lastActivity) > 30*time.Second {
return false
}

// 3. 检查错误计数
if conn.errorCount > 10 {
return false
}

return true
}

性能监控

性能统计

// cmd/rdma-stats.go

// RDMA 统计信息
type RDMAStats struct {
// 发送统计
TotalSentBytes int64 `json:"total_sent_bytes"`
TotalSentOps int64 `json:"total_sent_ops"`
AvgSendLatency int64 `json:"avg_send_latency"`
MaxSendLatency int64 `json:"max_send_latency"`

// 接收统计
TotalRecvBytes int64 `json:"total_recv_bytes"`
TotalRecvOps int64 `json:"total_recv_ops"`
AvgRecvLatency int64 `json:"avg_recv_latency"`
MaxRecvLatency int64 `json:"max_recv_latency"`

// 错误统计
TotalErrors int64 `json:"total_errors"`
SendErrors int64 `json:"send_errors"`
RecvErrors int64 `json:"recv_errors"`

// 连接统计
ActiveConnections int64 `json:"active_connections"`
TotalConnections int64 `json:"total_connections"`

// 内存统计
RegisteredRegions int64 `json:"registered_regions"`
TotalMemoryBytes int64 `json:"total_memory_bytes"`

// 时间戳
LastUpdate time.Time `json:"last_update"`
}

// 更新统计
func (s *RDMAStats) UpdateSendStats(bytes int64, latency int64) {
s.TotalSentBytes += bytes
s.TotalSentOps++

// 更新平均延迟
totalOps := s.TotalSentOps
s.AvgSendLatency = (s.AvgSendLatency*(totalOps–1) + latency) / totalOps

// 更新最大延迟
if latency > s.MaxSendLatency {
s.MaxSendLatency = latency
}

s.LastUpdate = time.Now()
}

func (s *RDMAStats) UpdateRecvStats(bytes int64, latency int64) {
s.TotalRecvBytes += bytes
s.TotalRecvOps++

totalOps := s.TotalRecvOps
s.AvgRecvLatency = (s.AvgRecvLatency*(totalOps–1) + latency) / totalOps

if latency > s.MaxRecvLatency {
s.MaxRecvLatency = latency
}

s.LastUpdate = time.Now()
}

// 获取性能报告
func (s *RDMAStats) GetReport() map[string]interface{} {
report := make(map[string]interface{})

report["total_sent_bytes"] = s.TotalSentBytes
report["total_recv_bytes"] = s.TotalRecvBytes
report["avg_send_latency_us"] = s.AvgSendLatency / 1000
report["avg_recv_latency_us"] = s.AvgRecvLatency / 1000
report["active_connections"] = s.ActiveConnections
report["total_errors"] = s.TotalErrors

throughput := float64(s.TotalSentBytes + s.TotalRecvBytes)
if !s.LastUpdate.IsZero() {
elapsed := time.Since(s.LastUpdate).Seconds()
if elapsed > 0 {
report["throughput_mbps"] = throughput / elapsed / 1024 / 1024 * 8
}
}

return report
}

小结

本文深入分析了 RDMA 传输层的实现：

RDMA 传输层是 MinIO 高性能的关键——让存储速度跟上网络带宽。

下节预告

下节预告：RDMA 内存注册——rdma-memory.go 中内存区域管理与 GPU-Direct 完整实现。

如果觉得本文对你有帮助，欢迎点赞、收藏、关注三连！ 本系列持续更新中，关注不迷路~