嵌入式 C 中的无锁环形缓冲区设计：单生产者-单消费者的内存序与 Cache 一致性保障

一、从互斥锁的隐性成本谈起：中断上下文中的阻塞等待

在嵌入式 C 开发中，环形缓冲区（Ring Buffer/FIFO）是最常见的跨上下文数据传输结构——串口接收中断向缓冲区写数据，主循环从缓冲区读数据。最简单的实现是全局数组加两个索引，在读写时用mutex_lock()保护临界区。然而当生产者位于中断上下文时，情况就不同了：

中断服务函数（ISR）中调用mutex_lock()可能导致死锁——如果锁已经被主循环持有且中断恰好发生在主循环的临界区内，ISR 将永远等待。FreeRTOS 提供了FromISR系列 API 来规避此问题，但它们依赖队列机制，引入了不必要的调度开销。无锁环形缓冲区通过精心设计的内存序（Memory Ordering）约束，在编译器和 CPU 重排的夹缝中保障正确性，是性能要求苛刻场景的最优解。

二、单生产者-单消费者（SPSC）的并发模型：为什么不需要锁

SPSC 场景不需要锁的数学直觉：生产者只写head索引，消费者只读head；消费者只写tail索引，生产者只读tail。两者没有写入竞争，读取竞争通过内存序约束解决。

sequenceDiagram participant PROD as 生产者（ISR / 线程） participant BUF as 环形缓冲区<br/>data[N] participant CONS as 消费者（主循环） Note over PROD,CONS: 初始状态：head=0, tail=0 PROD->>PROD: 1. 检查 head+1 != tail（空间充足？） alt 缓冲区已满 PROD->>PROD: 1a. 丢弃数据或覆盖旧数据 else 空间充足 PROD->>BUF: 2. 写入数据到 data[head] PROD->>PROD: 3. head = (head + 1) % N PROD->>PROD: 4. __sync_synchronize() 写屏障 end CONS->>CONS: A. 检查 head != tail（数据存在？） alt 缓冲区为空 CONS->>CONS: Aa. 休眠或返回空 else 有数据 CONS->>CONS: B. __sync_synchronize() 读屏障 CONS->>BUF: C. 读取 data[tail] CONS->>CONS: D. tail = (tail + 1) % N end

注意第 4 步的写屏障：它确保data[head] = value的写操作在head索引更新之前对所有观察者可见。没有这道屏障，消费者可能看到更新后的head但读到旧的（未写入的）data[head]，造成数据错乱。

三、生产级实现：内存屏障与 volatile 的正确使用

/* spsc_ringbuf.h —— 单生产者-单消费者无锁环形缓冲区 * * 设计约束： * - 仅支持 SPSC 场景（单生产者 + 单消费者） * - 不支持多生产者/多消费者（需改用 CAS 原子操作） * - 元素类型固定为 uint8_t，如需泛型请自行修改 * - 缓冲区大小 N 必须为 2 的幂（利用 & (N-1) 替代 % 运算） */ #ifndef SPSC_RINGBUF_H #define SPSC_RINGBUF_H #include <stdint.h> #include <stddef.h> #include <stdbool.h> // 编译器写屏障：阻止编译器对屏障前后的写操作重排 #if defined(__GNUC__) || defined(__clang__) #define COMPILER_BARRIER() __asm__ __volatile__("" ::: "memory") #else #error "不支持的编译器：请提供等效的编译器屏障宏" #endif // 全内存屏障：阻止 CPU 和编译器对屏障前后的内存操作重排 #if defined(__GNUC__) || defined(__clang__) #define MEMORY_BARRIER() __sync_synchronize() #else #error "不支持的编译器" #endif // 2 的幂取模 #define RING_MASK(size) ((size) - 1) typedef struct { uint8_t *buffer; // 数据缓冲区（外部分配） uint32_t size; // 缓冲区总大小（必须为 2 的幂） /* * 注意：head 和 tail 使用 volatile 并非为了解决并发竞争， * 而是防止编译器在循环读取时将其缓存到寄存器中。 * 真正的内存序保障由 MEMORY_BARRIER() 提供。 */ volatile uint32_t head; // 生产者写入位置 volatile uint32_t tail; // 消费者读取位置 } SpscRingBuf; // 初始化：外部提供 buffer 内存 static inline int ringbuf_init(SpscRingBuf *rb, uint8_t *buf, uint32_t size) { if (!rb || !buf || size == 0 || (size & (size - 1)) != 0) { return -1; // size 必须为 2 的幂 } rb->buffer = buf; rb->size = size; rb->head = 0; rb->tail = 0; return 0; } // 判断缓冲区已满 static inline bool ringbuf_is_full(SpscRingBuf *rb) { uint32_t head = rb->head; uint32_t tail = rb->tail; return ((head + 1) & RING_MASK(rb->size)) == tail; } // 判断缓冲区为空 static inline bool ringbuf_is_empty(SpscRingBuf *rb) { return rb->head == rb->tail; } /* * 生产者写入单个字节（可在 ISR 中调用） * * @return 实际写入字节数：成功=1, 缓冲区满=0 */ static inline int ringbuf_produce(SpscRingBuf *rb, uint8_t data) { uint32_t head = rb->head; uint32_t tail = rb->tail; // 消费者仅写 tail，此处读取是安全的 // 判断是否满（保留一个字节作为哨兵，简化空/满判断） if (((head + 1) & RING_MASK(rb->size)) == tail) { return 0; // 缓冲区已满，丢弃数据 } // ——— 关键区：先写数据，再更新 head ——— rb->buffer[head] = data; /* * 写屏障：确保 buffer[head] 的写入对所有观察者可见之后， * 才允许 head 的更新被其他核心/上下文看到。 * 如果没有此屏障，ARM 的 store buffer 可能将 head 更新 * 先于 buffer[head] 写入完成而对消费者可见。 */ MEMORY_BARRIER(); rb->head = (head + 1) & RING_MASK(rb->size); return 1; } /* * 消费者读取单个字节（在主循环或线程中调用） * * @param data 输出参数，读取的数据存放于此 * @return 实际读取字节数：成功=1, 缓冲区空=0 */ static inline int ringbuf_consume(SpscRingBuf *rb, uint8_t *data) { uint32_t head = rb->head; uint32_t tail = rb->tail; if (head == tail) { return 0; // 缓冲区为空 } /* * 读屏障：确保 head 的读取在当前读取 buffer[tail] 之前。 * 如果 head 读取时生产者尚未完成 buffer[head] 的写入， * 后续的 MEMORY_BARRIER 会阻止 buffer[tail] 的读取提前到 head 读取之前。 */ MEMORY_BARRIER(); *data = rb->buffer[tail]; /* * 这里不需要写屏障： * tail 的更新只需确保 buffer[tail] 已读取完毕。 * 消费者对 tail 的写入与生产者对 tail 的读取之间， * 生产者端的 MEMORY_BARRIER 已经建立了同步关系。 */ COMPILER_BARRIER(); rb->tail = (tail + 1) & RING_MASK(rb->size); return 1; } // 批量生产（适用 DMA 完成中断） static inline int ringbuf_produce_batch( SpscRingBuf *rb, const uint8_t *data, uint32_t len) { uint32_t head = rb->head; uint32_t tail = rb->tail; uint32_t mask = RING_MASK(rb->size); // 计算可用空间（保留一个哨兵） uint32_t free_space = (tail - head - 1) & mask; if (free_space < len) { len = free_space; // 截断至可用空间 if (len == 0) return 0; } // 分两段拷贝（环形缓冲区可能跨 buffer 尾部/头部边界） uint32_t first_chunk = rb->size - head; if (first_chunk >= len) { memcpy(&rb->buffer[head], data, len); } else { memcpy(&rb->buffer[head], data, first_chunk); memcpy(&rb->buffer[0], data + first_chunk, len - first_chunk); } MEMORY_BARRIER(); // 同单字节场景：先完成数据写入，再更新 head rb->head = (head + len) & mask; return len; } #endif /* SPSC_RINGBUF_H */

四、边界分析：ARM 弱内存序与 Cache 一致性的隐蔽陷阱

ARM 架构（Cortex-A/M）采用弱内存序模型（Weakly-Ordered Memory），仅依赖__sync_synchronize()（映射为DMB ISH指令）可能不够——它保证的是"屏障前的内存访问在屏障后的内存访问之前完成"，但不保证跨核心 Cache 的一致性。

Cache 一致性的三个场景：

1. 同核心 SPSC（ISR ↔ 主循环）：Cache 不存在一致性问题——两个上下文共享同一个核心的 L1 Cache。实践中最常见的场景，DMB足够。

2. 跨核心 SPSC（A53#0 ↔ A53#1）：数据可能暂存在发送方核心的 L1 Cache 中。MESI 协议的 invalidation 延迟最高可达数百周期。Cortex-A 系列需在数据写入后显式执行__builtin___clear_cache或依赖硬件 MOESI 协议的自动一致性。实践中跨核心场景建议使用stdatomic.h的memory_order_release/acquire代替手写屏障。

3. DMA 共访：DMA 引擎不经过 CPU Cache，直接读写物理内存。如果 CPU 侧对缓冲区的访问发生在 cached 地址，DMA 写入的数据可能被 stale cache line 覆盖。解决方案是使用 MMU 将 DMA 缓冲区域映射为non-cacheable。

性能数据（STM32H743 @ 480MHz）：

五、总结

SPSC 无锁环形缓冲区的正确性保障依赖于两个核心约束：生产者只写head、消费者只写tail的角色分离（消除写-写竞争），以及DMB写/读屏障确保数据写入在索引更新之前可见（消除读-写重排风险）。在 ISR ↔ 主循环的同核心场景下，该方案既避免了锁开销，又在架构上保证了正确性。跨核心场景建议从手写屏障迁移到 C11atomic+memory_order_release/acquire语义，利用编译器的自动屏障生成能力降低维护风险。DMA 共访场景则需额外关注 Cache 一致性问题——将缓冲区映射为 non-cacheable 或使用 cache flush/invalidate API。