队列状态机

flowchart TB N0[// queue.h: urpc_queue_status] N1[QUEUE_STATUS_IDLE // 空闲] N0 --> N1 N2[↓] N1 --> N2 N3[QUEUE_STATUS_RUNNING // 正在导入资源] N2 --> N3 N4[↓] N3 --> N4 N5[QUEUE_STATUS_READY // 就绪（可收发数据）] N4 --> N5 N6[↓] N5 --> N6 N7[QUEUE_STATUS_FAULT // 故障（可恢复）] N6 --> N7 N8[↓] N7 --> N8 N9[QUEUE_STATUS_ERR // 错误（不可恢复）] N8 --> N9

创建阶段

flowchart TB N0[queue_create_option_t *option] N1[1, sizeof(send_recv_queue_local_t] N2[provider, cfg_get, cfg, QUEUE_TRANS_MODE_JETTY] N3[provider, send_recv_local_q, cfg, qid] N4[中断模式] N5[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;
}

数据传输

flowchart TB N0[jetty, wr, &bad_wr] N1[jfs_jfc, ...] N2[jfr_jfc, ...]

排空与销毁

flowchart TB N0[queue_t *l_queue] N1[local] N2[l_queue] N3[local_q]

RX Buffer 管理

// 接收缓冲区上下文
typedef struct rx_buf_ctx {
    urpc_list_t node;
    umq_buf_t *buffer;
} rx_buf_ctx_t;

typedef struct rx_buf_ctx_list {
    void *addr;                             // 分配的内存起始地址
    urpc_list_t idle_rx_buf_ctx_list;       // 空闲列表
    urpc_list_t used_rx_buf_ctx_list;       // 已用列表
} rx_buf_ctx_list_t;

// 预填充 RX buffer（bind 后）
static int umq_ub_prefill_rx_buf(ub_queue_t *queue) {
    do {
        qbuf = umq_buf_alloc(queue->rx_buf_size, batch_count, 0, NULL);
        umq_ub_post_rx_inner_impl(queue, qbuf, &bad_buf);
        require_rx_count -= cur_batch_count;
    } while (require_rx_count > 0);
}