P1-03 · ring：免锁队列如何可能 对应路线图：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 与 RCUkfifo（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 数据结构"的实证。下一篇：P1-04 · mbuf：包的一生都住在里面