流控概述
UMQ UB 模式的流控是一个基于 credit 的窗口机制，使用 immediate 数据在带内交换窗口信息。
flowchart LR
subgraph Sender["Sender Side"]
TX["TX: send data"] --> FC_CHECK{"remote_rx_window
>= required?"} FC_CHECK -->|Yes| SEND["consume window
send data"] FC_CHECK -->|No| WAIT["wait for
window update"] WAIT --> READ_WIN["READ remote_win_buf"] READ_WIN --> TX end subgraph Receiver["Receiver Side"] RX["RX: receive data"] --> POST["local_rx_posted++"] POST --> CHECK_INT{"notify_interval
reached?"} CHECK_INT -->|Yes| NOTIFY["send IMM_TYPE_FLOW_CONTROL
with local_rx_posted"] CHECK_INT -->|No| RX end SEND -->|"UB Transport"| RX NOTIFY -->|"immediate data"| Sender style Sender fill:#e3f2fd,stroke:#163e7a style Receiver fill:#e8f5e9,stroke:#0f5d44
>= required?"} FC_CHECK -->|Yes| SEND["consume window
send data"] FC_CHECK -->|No| WAIT["wait for
window update"] WAIT --> READ_WIN["READ remote_win_buf"] READ_WIN --> TX end subgraph Receiver["Receiver Side"] RX["RX: receive data"] --> POST["local_rx_posted++"] POST --> CHECK_INT{"notify_interval
reached?"} CHECK_INT -->|Yes| NOTIFY["send IMM_TYPE_FLOW_CONTROL
with local_rx_posted"] CHECK_INT -->|No| RX end SEND -->|"UB Transport"| RX NOTIFY -->|"immediate data"| Sender style Sender fill:#e3f2fd,stroke:#163e7a style Receiver fill:#e8f5e9,stroke:#0f5d44
初始化与窗口建立
flowchart TB
N0[&queue->flow_control, info]
发送端：窗口消耗与检查
flowchart TB
N0[&fc, queue, 1]
N1[remote_rx_window >= required]
N2[fc, queue]
N3[READ, remote_win_buf_addr]
接收端：窗口通告
flowchart TB
N0[// umq_ub_fill_tx_imm 每次发送时：]
N2[// └─ 如果是，填充 IMM_TYPE_FLOW_CONTROL immediate：]
N1 --> N2
N3[// umq_ub_imm_t imm = .flow_control =]
N2 --> N3
N4[// .umq_private = 1,]
N3 --> N4
N5[// .type = IMM_TYPE_FLOW_CONTROL,]
N4 --> N5
N7[// ;]
N6 --> N7
原子 vs 非原子操作
流控窗口支持两种实现，通过 use_atomic_window 配置选择：
| 类型 | 实现 | 适用场景 |
|---|---|---|
| 非原子 | 直接读写 uint16_t | 单线程收发 |
| 原子 | __atomic_compare_exchange_n CAS | 多线程并发收发 |
// 原子操作实现（umq_ub_flow_control.c）
static inline uint16_t remote_rx_window_inc_atomic(struct ub_flow_control *fc, uint16_t new_win) {
uint16_t after, before = __atomic_load_n(&fc->remote_rx_window, __ATOMIC_RELAXED);
do {
win_sum = before + new_win;
if (win_sum > UINT16_MAX) { /* error */ break; }
after = (uint16_t)win_sum;
} while (!__atomic_compare_exchange_n(&fc->remote_rx_window, &before, after,
true, __ATOMIC_ACQ_REL, __ATOMIC_ACQUIRE));
return after;
}流控统计
| 统计项 | 含义 |
|---|---|
local_rx_posted | 当前本端已下发的 RX buffer 数 |
remote_rx_window | 对端通告的可用 RX 深度 |
total_local_rx_posted | 本端累计下发的 RX buffer 总数 |
total_local_rx_notified | 本端累计通告的可用深度总数 |
total_remote_rx_consumed | 发送端累计消耗的对端窗口总数 |
total_flow_controlled_wr | 因窗口不足失败的 WR 数 |