URPC Architecture Deep Dive: UMDK 统一远程过程调用协议架构分析

数据流向总览

                    ┌─────────────────────────────────────────────────────────────────┐
                    │                      数据流向图                                   │
                    │                                                                  │
                    │   Client                              Server                     │
                    │    │                                   │                        │
                    │    │  1. urpc_func_call()              │                        │
                    │    │     ├─ channel_get()              │                        │
                    │    │     ├─ RR select l_queue          │                        │
                    │    │     ├─ RR select r_queue          │                        │
                    │    │     ├─ req_entry_get()            │                        │
                    │    │     ├─ fill REQ header            │                        │
                    │    │     └─ queue->ops->send() ───────►│                        │
                    │    │                                   │  2. urma_poll_jfc()    │
                    │    │                                   │     └─ process_rx_cr() │
                    │    │                                   │         └─ func_exec() │
                    │    │                                   │            ↓           │
                    │    │                                   │     [用户函数执行]      │
                    │    │                                   │            ↓           │
                    │    │  4. poll REQ_RSPED               │     func_return()      │
                    │    │     ◄─────────────────────────────│         └─ send RSP   │
                    │    │                                   │                        │
                    └────┴───────────────────────────────────┴────────────────────────┘

                    ┌─────────────────────────────────────────────────────────────────┐
                    │                       UMQ 数据流向                                │
                    │                                                                  │
                    │   App                                    App                    │
                    │    │                                      │                     │
                    │    │  umq_enqueue()                       │                     │
                    │    │    └─ umq_tp_enqueue()               │                     │
                    │    │        └─ umq_ub_enqueue_impl()      │                     │
                    │    │            ├─ flow_control_check()   │                     │
                    │    │            ├─ fill_wr_impl()         │                     │
                    │    │            └─ urma_post() ──────────►│                     │
                    │    │                                      │  urma_poll_jfc()    │
                    │    │                                      │    └─ umq_dequeue()  │
                    │    │                                      │        └─ app       │
                    └────┴──────────────────────────────────────┴─────────────────────┘

模块关系与调用链

URPC 代码库包含约 60,000 行 C 代码，分为 URPC Framework 和 UMQ 两大子系统。理解模块间的调用关系是理解整个架构的关键。

模块依赖图

                         ┌─────────────────────────────────────────────┐
                         │           urpc_framework_api.h (公共 API)      │
                         │  urpc_init / urpc_func_call / urpc_func_poll  │
                         └──────────────────────┬──────────────────────┘
                                                │
            ┌───────────────────────────────────┼───────────────────────────────────┐
            │                                   │                                   │
            ▼                                   ▼                                   ▼
┌───────────────────────┐     ┌─────────────────────────────┐     ┌───────────────────────────┐
│   cp.c (控制面)        │     │   dp.c (数据面)              │     │   func.c (函数注册)         │
│   urpc_init()         │     │   urpc_func_call()         │     │   urpc_func_register()    │
│   urpc_server_start() │     │   urpc_func_poll()         │     │   urpc_func_exec()        │
│   channel 生命周期     │     │   urpc_func_return()       │     │   urpc_func_id_get()      │
└───────────┬───────────┘     └──────────┬──────────────────┘     └───────────────────────────┘
            │                            │
            ▼                            ▼
┌───────────────────────┐     ┌─────────────────────────────┐
│   channel.c (通道)     │     │   queue.c → queue_send_recv.c │
│   channel_alloc()     │     │   send_recv_create_local_queue()
│   req_entry_get()     │     │   send_recv_send()          │
│   server_node 管理     │     │   send_recv_post()          │
└───────────────────────┘     └──────────┬──────────────────┘
                                         │
                                         ▼
                              ┌─────────────────────────────┐
                              │  provider_ops_jetty.c        │
                              │  jetty_provider_init()      │
                              │  jetty_provider_register_mem()│
                              └──────────┬──────────────────┘
                                         │
                                         ▼
                              ┌─────────────────────────────┐
                              │  jetty_public_func.c         │
                              │  send_recv_create_jetty()   │
                              │  send_recv_poll_tx/rx()     │
                              └──────────┬──────────────────┘
                                         │
                                         ▼
                              ┌─────────────────────────────┐
                              │  URMA (urma_api.h)           │
                              │  urma_post_jetty_send_wr()  │
                              │  urma_poll_jfc()            │
                              └─────────────────────────────┘

                         ┌─────────────────────────────────────────────┐
                         │              umq_api.h (UMQ 公共 API)         │
                         │  umq_init / umq_create / umq_enqueue         │
                         └──────────────────────┬──────────────────────┘
                                                │
            ┌───────────────────────────────────┼───────────────────────────────────┐
            │                                   │                                   │
            ▼                                   ▼                                   ▼
┌───────────────────────┐     ┌─────────────────────────────┐     ┌───────────────────────────┐
│   umq_ub (UB 模式)     │     │   umq_ipc (IPC 模式)        │     │   umq_ubmm (共享内存)      │
│   umq_ub_impl.c       │     │   umq_ipc_impl.c           │     │   umq_ubmm_impl.c         │
│   umq_ub_buf_alloc   │     │   umq_ipc_post_impl        │     │   umq_ubmm_rendezvous     │
│   umq_ub_enqueue_impl│     │   umq_ipc_poll_impl        │     │   umq_ubmm_bind_impl      │
└───────────┬───────────┘     └─────────────────────────────┘     └───────────────────────────┘
            │
            ▼
┌───────────────────────┐
│  umq_ub/core/private/  │
│  umq_ub.c (核心实现)   │
│  umq_ub_bind_inner   │
│  umq_ub_post_rx/tx   │
│  umq_ub_read (UB_PLUS)│
└───────────┬───────────┘
            │
            ▼
┌───────────────────────┐
│  URMA (urma_api.h)     │
└───────────────────────┘

动态分发机制（多态）

URPC 大量使用函数指针实现多态，这是理解模块关系的关键。CodeGraph 揭示了以下动态分发链路：

// 三层动态分发（以 send 为例）：

// Layer 1: dp.c 通过 queue_ops 调用
// dp.c:499
errno = l_queue->ops->send(l_queue, &queue_wr);
//        ↑ 函数指针，运行时绑定

// Layer 2: queue_send_recv.c 设置 ops
// queue_send_recv.c:655
static queue_ops_t g_urpc_send_recv_ops = {
    .send = send_recv_send,
    .post = send_recv_post,
    .poll = send_recv_poll,
    .read = send_recv_read,
    .wait = send_recv_wait,
    ...
};

// Layer 3: jetty_public_func.c 实现
// jetty_public_func.c:865
int send_recv_send(queue_t *l_queue, queue_wr_t *wr) {
    // 最终调用 URMA
    urma_post_jetty_send_wr(jetty, urma_wr, &bad_wr);
}

// Layer 4: URMA 再次分发到硬件驱动
// urma_dp_api.c:278
urma_status_t urma_post_jetty_send_wr(urma_jetty_t *jetty, ...) {
    urma_ops_t *dp_ops = get_ops_by_urma_jetty(jetty);
    return dp_ops->post_jetty_send_wr(jetty, wr, bad_wr);
    //                  ↑ 硬件驱动函数指针
}

关键调用链（CodeGraph 验证）

以下是通过 CodeGraph 验证的 4 条核心调用链，展示了从用户 API 到硬件操作的完整路径：

核心模块函数表

以下是各核心模块的关键函数及其职责，按模块组织。每个模块先给出代码结构和核心数据结构，再列出关键函数。

cp.c — 控制面（3,014 行）

代码结构：cp.c 是 URPC Framework 的入口模块，负责系统初始化、角色管理和资源生命周期。

// 核心全局状态
static urpc_ctx_t g_urpc_ctx = {0};        // 全局 URPC 上下文
static urpc_ctrl_cb_t g_urpc_ctrl_msg_cb;  // 控制消息回调

// 初始化流程（按顺序）：
// urpc_init() → check_set_cfg() → async_event_ctx_init() → urpc_thread_ctx_init()
//   → dfx_init() → urpc_resource_release_init() → queue_id_allocator_init()
//   → provider_init() → urpc_server_client_init() → urpc_client_init()
//   → urpc_server_init() → urpc_state_update(URPC_STATE_INIT)

// 角色区分初始化：
//   ROLE_SERVER: server_keepalive_cfg_init() → server_channel_id_allocator_init()
//   ROLE_CLIENT: client_keepalive_cfg_init() → client_manage_channel_init()
//   ROLE_SERVER_CLIENT: 两者都执行

dp.c — 数据面（2,094 行）

代码结构：dp.c 处理所有数据面操作，包括函数调用、响应、轮询和错误处理。

// 核心数据结构
typedef struct tx_ctx {
    void *user_ctx;              // 用户上下文
    urpc_sge_t *sges;            // SGE 数组
    uint32_t sge_num;            // SGE 数量
    uint32_t req_id;             // 请求 ID
    uint8_t call_mode;           // 调用模式（EARLY_RSP/ACK/WAIT_RSP）
    uint8_t free_events_completed:1;  // 双次释放标志
    uint8_t sge_handover_completed:1; // SGE 已交给上层
    uint8_t internal_message:1;       // 内部消息标志
} tx_ctx_t;

// 扩展操作注册表（支持 256 个 func_defined 模块）
static ext_ops_t *g_urpc_ext_ops[MAX_FUNC_DEFINED];

// 完成事件类型（urpc_poll_event）
POLL_EVENT_REQ_ACKED      // 收到 ACK
POLL_EVENT_REQ_RSPED      // 收到 RSP
POLL_EVENT_REQ_ACKED_RSPED // 同时收到 ACK+RSP
POLL_EVENT_REQ_RECVED     // Server 收到 REQ
POLL_EVENT_REQ_SENDED     // REQ 发送完成
POLL_EVENT_RSP_SENDED     // RSP 发送完成
POLL_EVENT_READ_RET       // READ 完成
POLL_EVENT_ERR            // 错误事件

channel.c — 通道管理（1,471 行）

代码结构：channel.c 管理 Channel 的完整生命周期，包括请求条目、函数注册表和内存信息。

// 核心数据结构
typedef struct urpc_channel_info {
    uint32_t id;                         // Channel ID
    req_entry_t *req_entry_table;        // 请求条目表（hash + 线性探测）
    struct urpc_hmap func_tbl;           // 函数注册表
    urpc_list_t server_nodes_list;       // 连接的 Server 节点
    struct queue_nodes_head l_queue_nodes_head;  // 本地队列链表
    struct queue_nodes_head r_queue_nodes_head;  // 远端队列链表
    urpc_list_t mem_info_list;           // 导入的内存信息
    pthread_spinlock_t lock;             // 自旋锁（保护 req_entry 分配）
    pthread_rwlock_t rw_lock;            // 读写锁（数据面读，控制面写）
    uint32_t req_id;                     // 当前请求 ID（递增）
    uint32_t req_entry_size;             // 请求条目表大小（默认 1024）
} urpc_channel_info_t;

// 请求条目
typedef struct req_entry {
    pthread_mutex_t lock;                // 条目锁
    urpc_timer_t *timer;                 // 超时定时器
    urpc_sge_t *args;                    // 请求参数
    uint32_t req_id;                     // 请求 ID
    uint8_t valid;                       // 有效标志
    urpc_req_cb_t cb;                    // 完成回调
    void *cb_arg;                        // 回调参数
} req_entry_t;

// 全局 Channel 表
static urpc_channel_info_t *g_urpc_channels[URPC_MAX_CHANNELS];  // 最大 300,000

queue_send_recv.c — Jetty 队列（1,533 行）

代码结构：queue_send_recv.c 实现基于 Jetty 的队列操作，包括发送、接收、轮询和队列生命周期管理。

// 核心数据结构
typedef struct send_recv_queue_local {
    queue_local_t local_q;           // 基类（必须在前）
    urma_jetty_t *jetty;             // URMA Jetty 对象
    urma_jfc_t *jfs_jfc;             // 发送完成队列
    urma_jfc_t *jfr_jfc;             // 接收完成队列
    urma_jfce_t *jfce;               // 中断事件（可选）
    atomic_uint in_restore_process;  // 恢复过程标志
} send_recv_queue_local_t;

// 资源引用计数（共享资源管理）
typedef struct jfr_ctx {
    urma_jfr_t *jfr;                 // 接收队列
    urma_jfc_t *jfr_jfc;             // 接收完成队列
    urma_jfce_t *jfr_jfce;           // 接收中断事件
    volatile uint32_t ref_cnt;       // 引用计数
    rx_buf_ctx_list_t rx_buf_ctx_list; // RX buffer 上下文列表
} jfr_ctx_t;

// Queue ops 接口（多态分发）
static queue_ops_t g_send_recv_queue_ops = {
    .send = send_recv_send,
    .post = send_recv_post,
    .poll = send_recv_poll,
    .read = send_recv_read,
    .bind_queue = send_recv_bind_queue,
    .unbind_queue = send_recv_unbind_queue,
    .delete_local_queue = send_recv_delete_local_queue,
    .delete_remote_queue = send_recv_delete_remote_queue,
};

umq_ub.c — UMQ UB 模式（2,163 行）

代码结构：umq_ub.c 是 UMQ UB 模式的核心实现，处理绑定、发送、接收、流控和内存导入。

// 核心数据结构
typedef struct ub_queue {
    urma_jetty_t *jetty;             // Jetty 对象
    jfr_ctx_t *jfr_ctx;              // 接收上下文
    urma_jfc_t *jfs_jfc;             // 发送完成队列
    umq_ub_ctx_t *dev_ctx;           // 设备上下文
    struct ub_bind_ctx *bind_ctx;    // 绑定上下文
    ub_flow_control_t flow_control;  // 流控状态
    urma_target_seg_t **imported_tseg_list;  // 导入的远端 segment 列表
    umq_buf_t *notify_buf;           // 管理消息缓冲区
    uint32_t rx_depth, tx_depth;     // 队列深度
    umq_state_t state;               // 队列状态
} ub_queue_t;

typedef struct umq_ub_ctx {
    urma_context_t *urma_ctx;        // URMA 上下文
    urma_target_seg_t *tseg_list[UMQ_MAX_TSEG_NUM];  // 本地 segment 列表
    remote_imported_tseg_info_t *remote_imported_info;  // 远端导入信息
    umq_flow_control_cfg_t flow_control;  // 流控配置
} umq_ub_ctx_t;

// 远端 EID 管理
typedef struct remote_imported_tseg_info_t {
    bool tesg_imported[1024][255];   // 导入标志 [eid_id][tseg_id]
    urma_target_seg_t *imported_tseg_list[1024][255];  // 导入的 segment
    struct urpc_hmap remote_eid_id_table;  // EID → ID 映射
    util_id_allocator_t eid_id_allocator;  // ID 分配器
} remote_imported_tseg_info_t;

keepalive_task.c — 保活（922 行）

代码结构：keepalive_task.c 实现连接保活机制，使用定时器和可达列表检测连接健康状态。

// 核心数据结构
typedef struct urpc_keepalive_task_entry {
    pthread_spinlock_t lock;         // 条目锁
    urpc_instance_key_t key;         // 远端实例标识（EID + PID）
    struct urpc_hmap_node task_node; // 任务 hash 表节点
    uint64_t cpu_cycles;             // 最后活跃时间戳
    urpc_timer_t *timer;             // 超时定时器
    urpc_list_t list;                // 可达链表节点
    uint32_t server_chid[URPC_MAX_CHANNEL_PER_CLIENT];  // Server 通道 ID 列表
} urpc_keepalive_task_entry_t;

// 全局管理结构
static struct {
    pthread_rwlock_t lock;           // 读写锁
    struct urpc_hmap client_id_map;  // Client ID → entry
    struct urpc_hmap server_id_map;  // Server ID → entry
    struct urpc_hmap task_map;       // instance_key → entry
    struct urpc_hmap server_info_map;  // server_info → entry
    urpc_list_t reachable_list;      // 可达列表（逻辑 Server 更新）
} g_urpc_keepalive_mgmt;

// 配置参数
#define URPC_KEEPALIVE_TASK_NUM 8192   // 最大任务数
#define KEEPALIVE_MAX_PROB_CNT 127     // 最大探测次数
// keepalive_cycle_time: 发送间隔 [1, 3600]s
// keepalive_check_time: 超时判定 [cycle_time, 3600]s
// delay_release_time: 延迟释放 [1, 3600]s

crypto.c — 加密（~300 行）

代码结构：crypto.c 实现控制面 SSL/TLS 和数据面 AES-GCM 加密。

// 核心数据结构
typedef struct urpc_cipher {
    const EVP_CIPHER *cipher;        // EVP cipher 对象（AES-256-GCM）
    crypto_key_t crypto_key;         // 32 字节密钥
    uint32_t chid;                   // 通道 ID（IV 固定字段）
    atomic_ullong counter;           // IV 计数器（可变字段）
} urpc_cipher_t;

typedef struct crypto_key {
    uint8_t key[32];                 // 256-bit 密钥
} crypto_key_t;

// SSL 配置
typedef struct urpc_ssl_config {
    uint32_t ssl_flag;               // 启用/禁用标志
    urpc_ssl_mode_t ssl_mode;        // PSK / 证书
    urpc_tls_version_t min/max_tls_version;  // TLS 1.2 / 1.3
    char *cipher_list;               // TLS 1.2 cipher
    char *cipher_suites;             // TLS 1.3 cipher
    urpc_ssl_psk_client_cb_func client_cb_func;  // PSK 客户端回调
    urpc_ssl_psk_server_cb_func server_cb_func;  // PSK 服务端回调
} urpc_ssl_config_t;

// 加密头结构
typedef struct urpc_security_exthdr {
    uint8_t iv[12];                  // 初始化向量
    uint8_t tag[16];                 // GCM 认证标签
} urpc_security_exthdr_t;

UMQ 其他模式

umq_ipc（共享内存 IPC 模式，~1,000 行）：

// 核心数据结构
typedef struct umq_ipc_queue {
    int shm_fd;                      // 共享内存文件描述符
    void *shm_addr;                  // 映射地址
    umq_ipc_shm_hdr_t *shm_hdr;      // 共享内存头部（ring buffer 元数据）
    umq_ipc_ring_t *tx_ring;         // 发送 ring buffer
    umq_ipc_ring_t *rx_ring;         // 接收 ring buffer
    sem_t *tx_sem, *rx_sem;          // 进程间信号量
} umq_ipc_queue_t;

// 数据传输：通过共享内存 ring buffer，无 kernel 参与
// umq_ipc_post_impl() → 写入 tx_ring → sem_post(tx_sem)
// umq_ipc_poll_impl() → 等待 rx_sem → 从 rx_ring 读取

umq_ubmm（UB 共享内存模式，~850 行）：

// 核心数据结构
typedef struct umq_ubmm_queue {
    urma_target_seg_t *shm_seg;      // 共享内存 segment
    urma_ubva_t shm_ubva;            // UB 虚拟地址
    umq_ubmm_ring_t *tx_ring;        // 发送 ring
    umq_ubmm_ring_t *rx_ring;        // 接收 ring
} umq_ubmm_queue_t;

// 特点：同节点设备间通过 UB 共享内存零拷贝通信
// 使用 rendezvous 协议交换共享内存地址
// umq_ubmm_bind_impl() → 交换 UBVA → 直接读写共享内存

protocol.h — 协议定义（~600 行）

代码结构：protocol.h 定义 URPC 的所有协议消息格式和编解码函数。

// 消息类型
typedef enum urpc_msg_type {
    URPC_MSG_REQ,            // 请求
    URPC_MSG_ACK,            // 确认
    URPC_MSG_RSP,            // 响应
    URPC_MSG_ACK_AND_RSP,    // ACK+RSP 合并
    URPC_MSG_READ,           // 读取（参数拉取）
} urpc_msg_type_t;

// REQ 头（32B + DMA table）
struct urpc_req_head {
    uint64_t version : 4;
    uint64_t type : 4;
    uint64_t ack : 1;                // 是否需要 ACK
    uint64_t arg_dma_count : 5;      // DMA 地址表项数
    uint64_t function : 48;          // 函数 ID
    uint32_t req_total_size;         // 请求总大小
    uint32_t req_id;                 // 请求唯一 ID
    uint32_t client_channel : 24;    // 客户端通道 ID
    uint32_t function_defined : 8;   // 自定义字段
    arg_dma_t ub_dma[0];             // DMA 地址表（灵活数组）
};

// ACK 头（12B）
struct urpc_ack_head {
    uint8_t version : 4;
    uint8_t type : 4;
    uint16_t req_id_range;           // 请求 ID 范围
    uint32_t req_id;                 // 请求 ID
    uint32_t client_channel : 24;
};

// RSP 头（16B + offset array）
struct urpc_rsp_head {
    uint8_t version : 4;
    uint8_t type : 4;
    uint8_t status;                  // 执行状态
    uint16_t req_id_range;
    uint32_t req_id;
    uint32_t client_channel : 24;
    uint32_t function_defined : 8;
    uint32_t response_total_size;    // 响应总大小
    uint32_t return_data_offset[0];  // 返回数据偏移
};

// 控制面头
struct urpc_ctl_head {
    uint8_t version;
    uint8_t opcode;                  // URPC_CTL_*
    uint16_t error_code;
    uint16_t dp_encrypt : 1;         // 数据面加密标志
    uint16_t keepalive : 1;          // 保活标志
    uint32_t channel;                // 通道 ID
    uint32_t data_size;              // 负载大小
    uint32_t ctl_opcode;             // 操作码
    int task_id;                     // 任务 ID（异步）
};

Deep Dive 1: 完整 RPC 调用流

一次完整的 URPC 函数调用涉及 Client 端、Server 端、Worker 端的协同。下图展示了从 urpc_func_call() 到 urpc_func_return() 的完整代码路径。

Client 端：urpc_func_call()

// dp.c:352-533
uint64_t urpc_func_call(uint32_t chid, urpc_call_wr_t *wr, urpc_call_option_t *option) {
    // 1. 获取 channel（读锁保护）
    channel = channel_get(chid);

    // 2. 选择 local queue（round-robin）
    l_queue = urpc_get_local_queue(channel, option);
    //    └─ channel_get_next_local_queue() → RR 轮转

    // 3. 选择 remote queue（round-robin）
    r_queue = urpc_get_remote_queue(channel, option);

    // 4. 分配 tx_ctx（从 queue 的 slab allocator）
    ctx = queue_ctx_get(l_queue, QUEUE_CTX_TYPE_TX);

    // 5. 分配 req_entry（hash 表 + 线性探测）
    req_entry = req_entry_get(channel, ctx);
    //    └─ req_id = channel->req_id++ & (size-1)
    //    └─ while (req_entry->valid) { req_id++; linear_probe }

    // 6. 填充 REQ 协议头
    req_head = (urpc_req_head_t *)wr->args[0].addr;
    urpc_req_fill_basic_info(req_head, ack, channel_id);
    urpc_req_fill_req_info_without_dma(req_head, func_id, len, req_id, func_defined);

    // 7. 可选加密（keepalive 或 ext_ops->encrypt）
    func_call_sge_encrypt(func_defined, channel, r_queue, req_entry, wr);

    // 8. 填充 tx_ctx
    fill_req_tx_ctx(ctx, chid, l_qh, req_id, wr, option);

    // 9. 启动超时定时器（如果配置了 timeout）
    timeout_create(option, req_entry, func_defined, wr);

    // 10. 发送（调用 queue ops → send_recv_send → urma_post）
    l_queue->ops->send(l_queue, &queue_wr);

    return req_id;  // 返回请求句柄
}

Server 端：接收 REQ 与分发

// queue_send_recv.c: send_recv_poll_rx()
//   └─ urma_poll_jfc(jfr_jfc, cr_cnt, cr)  // 获取 RX 完成事件
//   └─ process_rx_cr(local_queue, msg, cr)

// dp.c: process_rx_cr() → 处理接收到的 REQ
//   └─ 解析 REQ 头：urpc_req_parse_function() / urpc_req_parse_req_id()
//   └─ 分配 req_ctx（从 queue 的 slab）
//   └─ 可选解密：crypto_decrypt_user_data()
//   └─ ext_ops->rx_req_process()（扩展处理）
//   └─ urpc_func_exec(func_id, args, args_sge_num, &rsps, &rsps_sge_num)
//       └─ 查找 handler = channel->func_tbl[func_id]
//       └─ handler->sync_handler(args, args_sge_num, ctx, &rsps, &rsps_sge_num)
//          【用户函数在此执行】

Server 端：返回 RSP

// dp.c: urpc_func_return()
int urpc_func_return(uint64_t urpc_qh, void *req_ctx, urpc_return_wr_t *wr, ...) {
    // 1. 获取 queue 和 req_ctx
    l_queue = (queue_t *)urpc_qh;
    ctx = (req_ctx_t *)req_ctx;

    // 2. 分配 tx_ctx
    tx_ctx = queue_ctx_get(l_queue, QUEUE_CTX_TYPE_TX);

    // 3. 填充 RSP 协议头
    rsp_head = (urpc_rsp_head_t *)rsps[0].addr;
    urpc_rsp_fill_basic_info(rsp_head, status, client_channel, ack);
    urpc_rsp_fill_one_req_info(rsp_head, req_id, response_total_size, option);

    // 4. 发送 RSP
    l_queue->ops->send(l_queue, &queue_wr);
}

Client 端：接收 ACK/RSP

// dp.c: urpc_func_poll() / urpc_func_poll_wait()
//   └─ 遍历 channel 的所有 local queue
//   └─ l_queue->ops->poll() → send_recv_poll_tx/rx()
//       └─ urma_poll_jfc(jfs_jfc, ...)  // TX CQE
//       └─ urma_poll_jfc(jfr_jfc, ...)  // RX CQE

// process_tx_msg() → process_req_cqe():
//   根据 call_mode 分支：
//   ├─ EARLY_RSP: 直接释放 tx_ctx，报告 REQ_RSPED
//   ├─ WAIT_RSP: 报告 REQ_SENDED，等待 RSP
//   └─ 其他: 解锁 req_entry，等待 RSP

// 收到 RSP 后：
//   └─ urpc_process_rsp_callback(req_entry, err_code)
//       └─ req_entry->cb(rsps, rsps_sge_num, err, arg, ctx)
//          【用户回调在此执行】

tx_ctx 生命周期（双次释放机制）

URPC 使用 tx_ctx_try_put() 实现双次释放：tx_ctx 必须经历两个"完成事件"才能被释放回 slab：

// dp.c:650-658
void tx_ctx_try_put(tx_ctx_t *ctx) {
    if (ctx->free_events_completed == FALSE) {
        ctx->free_events_completed = TRUE;  // 第一次：TX CQE
        return;
    }
    queue_ctx_put(QUEUE_CTX_TYPE_TX, ctx);  // 第二次：ACK/RSP → 真正释放
}

这解决了 TX CQE 和 ACK/RSP 并发到达时的竞态条件——无论哪个先到，都必须等两个都完成。

Deep Dive 2: 流控窗口状态机

UMQ UB 模式的流控是一个基于 credit 的窗口机制，使用 immediate 数据在带内交换窗口信息。

初始化与窗口建立

// umq_ub_flow_control.c: umq_ub_window_init()
// 当 local_rx_posted 首次达到 initial_window 时：
//   1. 将 initial_window 写入 notify_buf 的 OFFSET_FLOW_CONTROL 位置
//   2. 对端通过 READ 获取该值，初始化 remote_rx_window

// umq_ub_bind_inner_impl() 中：
//   └─ umq_ub_window_init(&queue->flow_control, info)
//       └─ 从 bind_info 读取 win_buf_addr → 设置 remote_win_buf_addr

发送端：窗口消耗与检查

// umq_ub_fill_wr_impl() 发送前：
//   └─ umq_ub_window_dec(&fc, queue, 1)  // 消耗 1 个窗口
//       └─ if (remote_rx_window >= required) → 扣减并返回
//       └─ else → 返回 0，wr 返回 -UMQ_ERR_EAGAIN

// 窗口不足时：
//   └─ umq_ub_window_read(fc, queue)  // 从对端 READ 最新窗口
//       └─ urma_post_jetty_send_wr(READ, remote_win_buf_addr)

接收端：窗口通告

// umq_ub_fill_tx_imm() 每次发送时：
//   └─ 检查 local_rx_posted - total_local_rx_notified >= notify_interval
//   └─ 如果是，填充 IMM_TYPE_FLOW_CONTROL immediate：
//       umq_ub_imm_t imm = { .flow_control = {
//           .umq_private = 1,
//           .type = IMM_TYPE_FLOW_CONTROL,
//           .window = local_rx_posted  // 通告本端可用深度
//       }};
//   └─ urma_wr->rw.notify_data = imm.value & UMQ_UB_IMM_WITHOUT_PRIVATE_BITS

接收端：处理窗口通告

// umq_ub_dequeue_impl() 收到数据时：
//   └─ 检查 immediate 的 umq_private bit
//   └─ 如果是 IMM_TYPE_FLOW_CONTROL：
//       └─ umq_ub_window_inc(&fc, imm.flow_control.window)
//           └─ remote_rx_window += new_win
//           └─ 如果之前因窗口不足阻塞，现在可以继续发送

流控统计

Deep Dive 3: UB_PLUS 反向拉取状态机

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

状态机

┌─────────────────────────────────────────────────────────────────────┐
│                     UB_PLUS 反向拉取协议                             │
│                                                                      │
│  [Sender]                    [Receiver]                              │
│     │                            │                                   │
│     │  1. REQ(IMM=REVERSE_PULL_MEM)│                                  │
│     │  ──────────────────────────► │                                  │
│     │    携带: addr, length,       │                                  │
│     │    msg_id, msg_num          │                                  │
│     │                            │  2. 解析元数据                     │
│     │                            │  3. umq_ub_read_ctx_create()      │
│     │                            │     预分配 dst_buf                 │
│     │                            │  4. 逐块发起 READ:                │
│     │                            │     for (i = 0; i < msg_num; i++) │
│     │                            │       urma_post(READ,              │
│     │                            │         src=ref_sge[i].addr,       │
│     │                            │         dst=local_buf)             │
│     │                            │                                   │
│     │  5. IMM=REVERSE_PULL_MEM_DONE                                 │
│     │  ◄────────────────────────── │                                  │
│     │                            │  6. 释放 dst_buf                   │
│     │  7. 继续后续处理            │                                   │
└─────────────────────────────────────────────────────────────────────┘

核心实现：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 级缓冲区：

// umq_ub_impl.c: umq_ub_plus_buf_alloc_impl()
if (buf_size < umq_buf_size_middle()) {
    // 小 buffer → 从普通 pool 分配
    umq_qbuf_alloc(request_size, ...);
} else {
    // 大 buffer → 从 huge pool 分配
    type = umq_huge_qbuf_get_type_by_size(buf_size);
    umq_huge_qbuf_alloc(request_size, ..., type);
}

Deep Dive 4: 队列生命周期

URPC 的队列经历创建 → 绑定 → 使用 → 销毁的完整生命周期。

创建阶段

// queue_send_recv.c: send_recv_create_local_queue()
queue_t *send_recv_create_local_queue(queue_create_option_t *option) {
    // 1. 分配 send_recv_queue_local_t
    send_recv_local_q = urpc_dbuf_calloc(1, sizeof(send_recv_queue_local_t));

    // 2. 设置配置（rx_buf_size, tx_depth, priority 等）
    send_recv_set_local_queue_normal_cfg(provider, cfg_get, cfg, QUEUE_TRANS_MODE_JETTY);

    // 3. 创建传输资源
    send_recv_create_transport_resource(provider, send_recv_local_q, cfg, qid);
    //    └─ send_recv_get_jfce_ctx() → urma_create_jfce() [中断模式]
    //    └─ send_recv_get_jfs_jfc_ctx() → urma_create_jfc()
    //    └─ send_recv_get_jfr_jfc_ctx() → urma_create_jfc()
    //    └─ send_recv_get_jfr_ctx() → urma_create_jfr()
    //    └─ send_recv_create_jetty() → urma_create_jetty()

    // 4. 初始化 queue 元数据
    send_recv_local_q_init(send_recv_local_q, provider, ops, flag, qid);
}

绑定阶段

// queue_send_recv.c: send_recv_bind_queue()
int send_recv_bind_queue(queue_t *l_queue, queue_t *r_queue) {
    // 1. 导入远端 jetty
    imported_queue->tjetty = urma_import_jetty(urma_ctx, &rjetty, &token);

    // 2. 绑定 jetty（建立连接）
    urma_bind_jetty(local_queue->jetty, imported_queue->tjetty);

    // 3. 记录远端 jetty_id
    local_queue->remote_jetty_id = imported_queue->tjetty->id;
    local_queue->is_binded = TRUE;
}

状态转换

// queue.h: urpc_queue_status
QUEUE_STATUS_IDLE → QUEUE_STATUS_RUNNING → QUEUE_STATUS_READY
                                              ↓
                              QUEUE_STATUS_FAULT → QUEUE_STATUS_ERR
                              QUEUE_STATUS_RESET → QUEUE_STATUS_ERR

// 销毁前必须排空（flush）：
// send_recv_delete_lq_validate() 检查：
//   ├─ tx_wr_cnt == 0（所有发送完成）
//   └─ rx_flush_done == TRUE 或 rx_wr_cnt == 0（所有接收完成）

销毁阶段

// queue_send_recv.c: send_recv_delete_local_queue()
int send_recv_delete_local_queue(queue_t *l_queue) {
    // 1. 排空检查
    send_recv_delete_lq_validate(local);

    // 2. 解绑
    l_queue->ops->unbind_queue(l_queue);  // urma_unbind_jetty()

    // 3. 清理 slab
    queue_slab_uninit(local_q);

    // 4. 销毁传输资源
    send_recv_destroy_transport_resource(local);
    //    └─ urma_delete_jetty()
    //    └─ send_recv_put_jfr_ctx() → urma_delete_jfr()
    //    └─ send_recv_put_jfc_ctx() → urma_delete_jfc()
    //    └─ send_recv_put_jfce_ctx() → urma_delete_jfce()
}

Deep Dive 5: 内存导入协议

UMQ 的多内存池导入协议允许一个连接导入多个远端内存池，支持动态扩展。

初始化交换

// umq_ub_bind_info_get_impl() 构造 bind_info：
info->tseg = queue->dev_ctx->tseg_list[UMQ_QBUF_DEFAULT_MEMPOOL_ID];  // 默认 pool
info->win_buf_addr = umq_ub_notify_buf_addr_get(queue, OFFSET_FLOW_CONTROL);
info->notify_buf = umq_ub_notify_buf_addr_get(queue, OFFSET_MEM_IMPORT);

// 对端收到 bind_info 后：
// umq_ub_bind_inner_impl() → umq_ub_eid_id_get()
//   └─ 检查 remote_eid_id_table（hash map）
//   └─ 如果新 EID → 分配 eid_id → import_mem() → 插入 hmap

多内存池导入

// umq_ub.c: umq_ub_data_plan_import_mem()
int umq_ub_data_plan_import_mem(uint64_t umqh_tp, umq_buf_t *rx_buf, uint32_t ref_seg_num) {
    // 1. 解析接收缓冲区头部的 umq_imm_head
    umq_imm_head_t *umq_imm_head = (umq_imm_head_t *)rx_buf->buf_data;

    // 2. 解析 import_mempool_info 数组
    ub_import_mempool_info_t *import_mempool_info =
        (ub_import_mempool_info_t *)(rx_buf->buf_data + sizeof(umq_imm_head_t) + ref_seg_num * sizeof(ub_ref_sge_t));

    // 3. 逐个导入内存池
    for (uint32_t i = 0; i < umq_imm_head->mempool_num; i++) {
        // 检查是否已导入
        if (queue->imported_tseg_list[import_mempool_info[i].mempool_id] != NULL) {
            umq_ub_import_mem_done(queue, import_mempool_info[i].mempool_id);
            continue;
        }

        // 导入内存
        xchg_mem_info_t mem_info = { ... };
        imported_tseg = import_mem(queue->dev_ctx->urma_ctx, &mem_info);

        // 发送 IMPORT_DONE 确认
        umq_ub_import_mem_done(queue, import_mempool_info[i].mempool_id);
        //    └─ umq_ub_write_imm(IMM_TYPE_MEM_IMPORT_DONE)
    }
}

内存查找路径

发送数据时，根据 mempool_id 查找已导入的远端内存：

// umq_ub_fill_wr_impl() 发送时：
//   └─ buf_pro->mempool_id → 确定使用哪个内存池
//   └─ queue->imported_tseg_list[mempool_id] → 获取远端 tseg
//   └─ urma_sge.tseg = queue->imported_tseg_list[mempool_id]
//   └─ urma_post_jetty_send_wr() → 使用 tseg 发送

Deep Dive 6: Keepalive 保活机制

URPC 的保活机制使用专用队列和定时器实现连接健康检测。

架构

┌─────────────────────────────────────────────────────────────┐
│                    Keepalive 架构                             │
│                                                              │
│  ┌──────────────┐    ┌──────────────┐    ┌──────────────┐   │
│  │ Client Side  │    │  Timer Wheel │    │ Server Side  │   │
│  │              │    │              │    │              │   │
│  │ keepalive_   │───►│ urpc_timer_  │◄───│ keepalive_   │   │
│  │ task_entry   │    │ start/check  │    │ task_entry   │   │
│  └──────────────┘    └──────────────┘    └──────────────┘   │
│         │                                       │            │
│         ▼                                       ▼            │
│  ┌──────────────┐                      ┌──────────────┐     │
│  │ 发送 REQ     │                      │ 接收 REQ     │     │
│  │ (keepalive   │─────────────────────►│ 更新 ts      │     │
│  │  function_id)│                      │ 回复 RSP     │     │
│  └──────────────┘                      └──────────────┘     │
└─────────────────────────────────────────────────────────────┘

定时器检查

// keepalive_task.c: urpc_keepalive_check()
void urpc_keepalive_check(void *args) {
    // 遍历 reachable_list
    URPC_LIST_FOR_EACH(entry, list, &g_urpc_keepalive_mgmt.reachable_list) {
        // 检查是否超时
        if (current_time - entry->timestamp > check_time) {
            // 超时 → 回调通知
            keepalive_callback(URPC_KEEPALIVE_FAILED, info);
            // 延迟释放
            entry->timer = urpc_timer_start(delay_release_time, ...);
        } else {
            // 未超时 → 更新 timer
            urpc_timer_start(entry->timer, check_time, urpc_keepalive_check, ...);
        }
    }
}

保活消息格式

保活消息复用 URPC REQ 头格式，使用专用 function ID：

// protocol.h:
#define URPC_KEEPALIVE_FUNCTION_ID  (0x002001000005)
#define URPC_KEEPALIVE_HDR_SIZE     (sizeof(urpc_req_head_t) + sizeof(urpc_keepalive_head_t))

// keepalive_head 结构：
typedef struct urpc_keepalive_head {
    uint8_t version : 4;
    uint8_t is_rsp : 1;          // 是否为响应
    uint8_t status;              // 处理结果
    uint16_t l_qid;              // 本地队列 ID
    uint32_t server_channel : 24;
    uint32_t rsvd[9];            // 最后 28B 用于加密头
} urpc_keepalive_head_t;

Deep Dive 7: 控制面 IP 传输层

URPC 控制面使用 TCP/IP 连接传输控制消息（attach/detach/keepalive 等）。

传输层架构

// transport.c: 控制面 IP 传输
//   ├─ Server 端：监听 TCP 端口 → 接受连接 → SSL 握手 → IO 事件处理
//   └─ Client 端：发起 TCP 连接 → SSL 握手 → 发送控制消息

// 关键数据结构：
urpc_client_connect_hamp    // Client 连接 hash table（key: server_inner）
urpc_server_accept_manager  // Server 连接管理
server_resource_hamp        // Server 资源 hash table（key: client_key）

SSL 握手流程

// transport.c: client_do_ssl_handshake()
//   1. crypto_ssl_init(sockfd, false)  // 创建 SSL 对象
//   2. crypto_ssl_connect(ssl, &err)    // 发起连接
//   3. 注册 epoll 事件：client_on_ssl_handshake()
//   4. 握手完成 → client_on_io_event_process()

// transport.c: server_do_ssl_handshake()
//   1. crypto_ssl_init(sockfd, true)   // 创建 SSL 对象
//   2. crypto_ssl_accept(ssl, &err)    // 接受连接
//   3. 注册 epoll 事件：server_on_ssl_handshake()
//   4. 握手完成 → server_on_io_event_process()

控制消息处理

// transport.c: client_on_io_event_process()
//   └─ SSL_read() 读取消息
//   └─ 解析 urpc_ctl_head（version/opcode/channel/data_size）
//   └─ urpc_ctrl_msg_process(msg_type, ctrl_msg)
//       └─ 根据 opcode 分发：
//           ├─ URPC_CTL_SESSION_UPDATE → 更新会话
//           ├─ URPC_CTL_QUEUE_INFO_BIND → 绑定队列
//           ├─ URPC_CTL_QUEUE_INFO_ADD → 添加队列
//           └─ ...

安全加密与扩展机制

SSL/TLS 控制面加密

URPC 支持基于 OpenSSL 的控制面加密：

• 通过 urpc_ssl_config_set 配置 SSL 参数（PSK 模式或证书模式）
• 支持 TLS 1.2 / TLS 1.3
• 可配置 cipher list 和 cipher suites

数据面 AES-GCM 加密

// crypto.c: 数据面加密上下文
typedef struct urpc_cipher {
    const EVP_CIPHER *cipher;          // AES-256-GCM
    crypto_key_t crypto_key;           // 32B 密钥
    uint32_t chid;                     // 通道 ID（IV 固定字段）
    atomic_ullong counter;             // IV 计数器
} urpc_cipher_t;

// 加密流程：
// crypto_encrypt_user_data(ctx, sge, sge_num, hdr_index, first_sge_offset)
//   └─ EVP_CIPHER_CTX_init(ctx)
//   └─ EVP_EncryptInit_ex(ctx, EVP_aes_256_gcm(), NULL, key, iv)
//   └─ EVP_EncryptUpdate() → 加密 SGE 数据
//   └─ EVP_EncryptFinal_ex() → 完成加密
//   └─ EVP_CIPHER_CTX_ctrl(EVP_CTRL_GCM_GET_TAG, 16, tag) → 获取认证标签

扩展机制（Ext Ops）

URPC 提供 ext_ops_t 扩展接口，允许用户自定义：

// dp.h: ext_ops_t
typedef struct ext_ops {
    int (*rx_req_process)(ext_call_ctx_t *, req_ctx_t **, urpc_poll_msg_t *);
    int (*tx_read_process)(ext_call_ctx_t *, urpc_poll_msg_t *);
    int (*tx_read_err_process)(ext_call_ctx_t *, urpc_poll_msg_t *);
    void (*ext_proto_fill_extra_info)(urpc_req_head_t *, queue_local_t *, queue_remote_t *);
    bool (*timeout_create_process)(urpc_call_wr_t *, req_entry_t *);
    int (*encrypt)(urpc_hdr_type_t, urpc_cipher_t *, urpc_sge_t *, uint32_t);
    int (*decrypt)(urpc_hdr_type_t, urpc_cipher_t *, urpc_sge_t *, uint32_t);
    int (*stream_send_process)(void *, uint32_t, uint32_t, uint32_t);
    int (*rx_rsp_process)(ext_call_ctx_t *, urpc_poll_msg_t *);
} ext_ops_t;

Summary Table: URPC 在 UB 上的增量功能总览

Conclusion

URPC 在 UB 总线上构建了一套完整的"从硬件原语到函数调用语义"的技术栈。通过本次深度代码分析，我们揭示了以下关键洞察：

• 分层抽象：URMA → UMQ/URPC Framework → 公共 API，每层只关注本层职责，通过 ops 接口实现多态
• 带内控制：利用 immediate 数据编码流控、内存导入、反向拉取等控制协议，不占用数据通道
• 并发安全：tx_ctx 双次释放、req_entry hash+线性探测、读写锁分离等精细并发控制
• 用户可选：每一层增量功能都是可选的（feature flag），适应不同场景
• 生产级特性：Keepalive、DFX、SSL/TLS、300K 并发 Channel

从架构视角看，URPC 的设计遵循了"分层抽象、带内控制、用户可选"的原则——既能以最小开销运行在延迟敏感的推理场景，也能以全功能模式运行在需要流控和加密的训练/存储场景。

📄 详细分析文档（独立阅读）

以下文档将本报告的各个章节拆分为独立的 HTML 文件，方便单独阅读和分享。每个文档都包含完整的 Mermaid 可视化图表。

References

• UMDK README — umdk/README.md、umdk/README.ch.md
• URPC Framework API — umdk/src/urpc/include/framework/urpc_framework_api.h
• URPC Framework Types — umdk/src/urpc/include/framework/urpc_framework_types.h
• UMQ API — umdk/src/urpc/include/umq/umq_api.h
• UMQ Types — umdk/src/urpc/include/umq/umq_types.h
• URPC Protocol — umdk/src/urpc/framework/protocol/protocol.h
• URPC Datapath — umdk/src/urpc/framework/lib/datapath/dp.c（2,094 行）
• URPC Control Plane — umdk/src/urpc/framework/lib/control/cp.c（3,014 行）
• Channel 管理 — umdk/src/urpc/framework/core/channel/channel.c（1,471 行）
• Queue Send/Recv — umdk/src/urpc/framework/core/queue/jetty/queue_send_recv.c（1,533 行）
• Jetty Provider — umdk/src/urpc/framework/core/queue/jetty/provider_ops_jetty.c
• Jetty Public Func — umdk/src/urpc/framework/core/queue/jetty/jetty_public_func.c（1,533 行）
• UMQ UB Core — umdk/src/urpc/umq/umq_ub/core/private/umq_ub.c（2,163 行）
• UMQ UB Impl — umdk/src/urpc/umq/umq_ub/core/umq_ub_impl.c（1,485 行）
• UMQ UB Flow Control — umdk/src/urpc/umq/umq_ub/core/flow_control/umq_ub_flow_control.c
• UMQ UB Imm Data — umdk/src/urpc/umq/umq_ub/core/umq_ub_imm_data.h
• Keepalive Task — umdk/src/urpc/framework/lib/manager/keepalive_task.c（922 行）
• Keepalive Probe — umdk/src/urpc/framework/lib/manager/keepalive_probe.c
• Crypto — umdk/src/urpc/framework/lib/control/crypto/crypto.c
• Control Plane IP Transport — umdk/src/urpc/framework/lib/control/ip/transport.c（1,548 行）
• UMDK 中文文档 — umdk/doc/ch/urpc/