对应路线图：Phase 1 (Day 22-49) ｜ DPDK 24.11.6 LTS ｜ 内核对照 v6.6
代码出处：source/dpdk-stable-24.11.6/lib/ring/，引用均标文件:行号
上一篇：P1-02 mempool ｜ 下一篇：P1-04 mbuf
一句话定位
rte_ring 是 DPDK 的"万能胶"：mempool 的默认后端是它，进程间通信用它，P3 的 LB 分发用它。理解 ring = 理解 CAS 抢占 + tail 有序发布 这套无锁队列范式。
真实结构（lib/ring/rte_ring_core.h）
/** prod/cons sync types */
enum rte_ring_sync_type { // rte_ring_core.h:53
RTE_RING_SYNC_MT, /**< multi-thread safe (default mode) */
RTE_RING_SYNC_ST, /**< single thread only */
RTE_RING_SYNC_MT_RTS, /**< multi-thread relaxed tail sync */
RTE_RING_SYNC_MT_HTS, /**< multi-thread head/tail sync */
};
struct rte_ring_headtail { // rte_ring_core.h:65
volatile RTE_ATOMIC(uint32_t) head; /**< prod/consumer head. */
volatile RTE_ATOMIC(uint32_t) tail; /**< prod/consumer tail. */
union {
enum rte_ring_sync_type sync_type;
uint32_t single;
};
};
struct rte_ring { // rte_ring_core.h:116
alignas(RTE_CACHE_LINE_SIZE) char name[RTE_RING_NAMESIZE];
int flags;
const struct rte_memzone *memzone;
uint32_t size; /**< Size of ring. */
uint32_t mask; /**< Mask (size-1) of ring. */
uint32_t capacity; /**< Usable size of ring */
RTE_CACHE_GUARD;
/** Ring producer status. */
union __rte_cache_aligned {
struct rte_ring_headtail prod;
struct rte_ring_hts_headtail hts_prod;
struct rte_ring_rts_headtail rts_prod;
};
RTE_CACHE_GUARD;
/** Ring consumer status. */
union __rte_cache_aligned {
struct rte_ring_headtail cons;
struct rte_ring_hts_headtail hts_cons;
struct rte_ring_rts_headtail rts_cons;
};
RTE_CACHE_GUARD;
};结构注释里藏着核心设计（rte_ring_core.h:106-115）：
head/tail 不是 0..size-1 之间的值，而是 0..2^32-1，访问 ring[] 时才用 mask。因此两个 uint32 索引的减法在模 32 位意义下永远正确，溢出不是问题。布局要点：prod 和 cons 各占独立 cache line（RTE_CACHE_GUARD 隔离）——生产者只写 prod 行，消费者只写 cons 行，双方互不挤占对方的缓存行。size 必须 2 的幂 → & mask 替代 % size。
入队四步：对真实代码逐行讲
第 ① 步 · 抢占区间（lib/ring/rte_ring_generic_pvt.h:55-101，节选）：
static __rte_always_inline unsigned int
__rte_ring_move_prod_head(struct rte_ring *r, unsigned int is_sp,
unsigned int n, enum rte_ring_queue_behavior behavior,
uint32_t *old_head, uint32_t *new_head,
uint32_t *free_entries)
{
const uint32_t capacity = r->capacity;
// ...
do {
n = max;
*old_head = r->prod.head;
/* add rmb barrier to avoid load/load reorder in weak
* memory model. It is noop on x86
*/
rte_smp_rmb(); // :74
/*
* The subtraction is done between two unsigned 32bits value
* (the result is always modulo 32 bits ...). So 'free_entries'
* is always between 0 and capacity (which is < size).
*/
*free_entries = (capacity + r->cons.tail - *old_head); // :82
if (unlikely(n > *free_entries))
n = (behavior == RTE_RING_QUEUE_FIXED) ?
0 : *free_entries;
if (n == 0)
return 0;
*new_head = *old_head + n;
if (is_sp) {
r->prod.head = *new_head; // SP：普通写即可
success = 1;
} else
success = rte_atomic32_cmpset((uint32_t *)&r->prod.head,
*old_head, *new_head); // MP：CAS 抢区间
} while (unlikely(success == 0));
return n;
}第 ② 步 · 写入对象：__rte_ring_enqueue_elems（rte_ring_elem_pvt.h:129-152）按元素大小分派 32/64/128 字节三档批量复制——定长批量 copy，编译器可向量化。
第 ③④ 步 · 屏障 + 发布 tail（rte_ring_generic_pvt.h:13-30，全文）：
static __rte_always_inline void
__rte_ring_update_tail(struct rte_ring_headtail *ht, uint32_t old_val,
uint32_t new_val, uint32_t single, uint32_t enqueue)
{
if (enqueue)
rte_smp_wmb();
else
rte_smp_rmb();
/*
* If there are other enqueues/dequeues in progress that preceded us,
* we need to wait for them to complete
*/
if (!single)
rte_wait_until_equal_32((volatile uint32_t *)&ht->tail, old_val,
rte_memory_order_relaxed);
ht->tail = new_val;
}两个知识点：
- wmb/rmb 的选择：入队前
wmb（对象内容先于 tail 发布可见），出队前rmb（读到发布后的 tail 才读对象）。 - MP 场景的按序发布：
rte_wait_until_equal_32等前一个生产者的 tail 到old_val再写自己的——即使 P2 先写完数据，tail 也只能按序发布，消费者看到的永远是有序完整的数据。这就是"head 无锁抢占、tail 顺序发布"的完整语义。
入队主循环（rte_ring_elem_pvt.h:320-341，全文）：
static __rte_always_inline unsigned int
__rte_ring_do_enqueue_elem(struct rte_ring *r, const void *obj_table,
unsigned int esize, unsigned int n,
enum rte_ring_queue_behavior behavior, unsigned int is_sp,
unsigned int *free_space)
{
uint32_t prod_head, prod_next;
uint32_t free_entries;
n = __rte_ring_move_prod_head(r, is_sp, n, behavior,
&prod_head, &prod_next, &free_entries);
if (n == 0)
goto end;
__rte_ring_enqueue_elems(r, prod_head, obj_table, esize, n);
__rte_ring_update_tail(&r->prod, prod_head, prod_next, is_sp, 1);
end:
// ...
return n;
}出队是完美镜像（rte_ring_elem_pvt.h:367-389）：move_cons_head → dequeue_elems → update_tail(cons, ..., 0)。
变体一句话
- RTS（Relaxed Tail Sync）：head/tail 以
(cnt, pos)打包成 64 位原子（rte_ring_core.h:76-83），生产者可提前发布，靠 watermark 背压——用容量利用率换更少的缓存行竞争。 - HTS（Head/Tail Sync）：head/tail 打包进一个 64 位原子（
rte_ring_core.h:92-104），一次原子操作同时读两个值。 - 两阶段 peek API：
rte_ring_enqueue_peek占位 → 逐包填充（填不下就丢）→rte_ring_enqueue_do_commit发布，适合网卡收包场景。
🐧 内核对照：kfifo 与 RCU
kfifo（lib/kfifo.c） 是内核最接近 ring 的通用无锁 FIFO，屏障用法一一对应：
static void kfifo_copy_in(struct __kfifo *fifo, const void *src,
unsigned int len, unsigned int off) // kfifo.c:88
{
// ... memcpy 两段（处理回绕） ...
/*
* make sure that the data in the fifo is up to date before
* incrementing the fifo->in index counter
*/
smp_wmb(); // kfifo.c:110
}
unsigned int __kfifo_in(struct __kfifo *fifo,
const void *buf, unsigned int len) // kfifo.c:113
{
unsigned int l;
l = kfifo_unused(fifo);
if (len > l)
len = l;
kfifo_copy_in(fifo, buf, len, fifo->in);
fifo->in += len; // "发布"
return len;
}出队侧 kfifo_copy_out 同样在 out 前 smp_wmb()（kfifo.c:145-150）。与 ring 的差异：kfifo 是 SPSC 假设（in/out 各一人写），ring 用 CAS 把同样的屏障范式扩展到了 MP/MC。
RCU（kernel/rcu/）：另一种"免锁"哲学——读方零开销，写方延迟回收。ring 的免锁是队列语义（生产者-消费者互不阻塞）；RCU 的免锁是发布-订阅语义（读者随时读，写者保证读者不踩到已释放内存）。面试能讲出这个区别是加分项。
面试 3 问
- 画 4 线程并发 enqueue 时序图，tail 最后是多少？ → 各线程 CAS 抢到互不重叠区间；tail 按
wait_until_equal_32顺序推进，最终 = 实际入队总数。 - 少一条 wmb 会怎样？ → 消费者可能先看到 tail 发布再读到未写完的对象（数据竞争），且只在多核乱序窗口偶现，极难复现——所以源码注释必须写清屏障"保护什么"。
- head/tail 为什么用 uint32 且允许回绕？ → 模 32 位减法天然正确（
rte_ring_core.h:106-115注释），省掉回绕判断。
🔧 动手实验
扩展 code/ring-bench/：SPSC/MPSC/MPMC × 1/2/4/8 producer 吞吐+延迟矩阵。预期曲线：SPSC 最快（无 CAS 无等待），MPMC 在 8 生产者时见顶——数据留档 perf-data/P1/，这就是"热路径要 per-core 数据结构"的实证。
评论 (0)