06 UB_PLUS 反向拉取 Deep Dive

UB_PLUS 概述

UB 原生最大消息 64KB，UB_PLUS 通过"反向拉取"机制扩展到 10MB。这是一个多步握手协议。

状态机

核心实现：umq_ub_read()

// umq_ub.c:1254-1310
int umq_ub_read(uint64_t umqh_tp, umq_buf_t *rx_buf, umq_ub_imm_t imm) {
    // 1. 解析 immediate 中的元数据
    uint16_t buf_num = imm.ub_plus.msg_num;    // 批处理 buf 数量
    uint16_t msg_id = imm.ub_plus.msg_id;      // 消息 ID（用于去重）

    // 2. 解析接收缓冲区头部的 umq_imm_head
    umq_imm_head_t *umq_imm_head = (umq_imm_head_t *)rx_buf->buf_data;
    ub_ref_sge_t *ref_sge = (ub_ref_sge_t *)(umq_imm_head + 1);

    // 3. 预分配本地接收缓冲区
    umq_buf_t *ctx_buf = umq_ub_read_ctx_create(queue, umq_imm_head, buf_num, msg_id);
    //    └─ 根据 mem_interval 计算总大小
    //    └─ umq_buf_alloc(total_size, 1, ...) 从 huge pool 分配

    // 4. 逐块发起 READ
    for (uint32_t i = 0; i < buf_num; i++) {
        src_sge[i].addr = ref_sge[i].addr;           // 远端地址
        src_sge[i].tseg = imported_tseg_list[ref_sge[i].mempool_id];
        dst_sge[i].addr = tmp_buf->buf_data;          // 本地地址
        dst_sge[i].tseg = tseg_list[tmp_buf->mempool_id];
        urma_post_jetty_send_wr(queue->jetty, &urma_wr, &bad_wr); // READ
    }

    // 5. 完成后发送 REVERSE_PULL_MEM_DONE 确认
    //    └─ umq_ub_write_imm(imm = REVERSE_PULL_MEM_DONE)
}

缓冲区分配策略

UB_PLUS 使用独立的 huge qbuf pool，支持 8MB 级缓冲区：

Immediate 子类型

// umq_ub_imm_data.h: UB_PLUS 子类型
typedef enum umq_ub_plus_imm_sub_type {
    IMM_TYPE_UB_PLUS_DEFAULT,           // 默认子类型
    IMM_TYPE_REVERSE_PULL_MEM,          // 请求反向拉取
    IMM_TYPE_REVERSE_PULL_MEM_FREE,     // 释放反向拉取内存
    IMM_TYPE_REVERSE_PULL_MEM_DONE,     // 反向拉取完成确认
    IMM_TYPE_UB_PLUS_MAX
} umq_ub_plus_imm_sub_type_t;