SPSC 无锁 FIFO：2^n 大小的单生产者单消费者环形缓冲区

在嵌入式开发中，生产者/消费者是最常见的一类问题：串口中断收数据、DMA 搬运、音频流、协议解析……总是一边往里写、一边从里读。很多人第一反应是加互斥锁，但在「一个生产者、一个消费者」（SPSC，Single Producer Single Consumer）的场景下，锁完全是多余的——用对方法，不但不用加锁，还能做到零等待、无惊群、几乎零开销。

本文给出一个经过实战检验的 SPSC 无锁环形缓冲区（Ring Buffer）实现，核心只有两点：容量取 2 的幂（2^n），以及生产/消费各只改自己的索引。文末附可直接编译运行的 demo 和中断场景用法。

一、为什么容量必须是 2^n

环形缓冲区要在一维数组上"绕圈"。最常见的麻烦是「满和空都用 head==tail 表示，产生歧义」，于是大家被迫空出一个槽（用 (head+1)%size==tail 判满），既浪费一个字节又多一次取模。

2^n 大小可以一并解决这两个问题：

取模变位与：定位下标用 index & (SIZE-1)，编译器一条 and 指令搞定，比 % SIZE 快得多，且没有除法。
索引自由递增，天然回绕：让 write_idx / read_idx 一直往上加（uint32_t 到 42 亿后自然回绕），用 write_idx - read_idx 直接得到"已写入未读取"的字节数。可用量 = 索引差，等于 SIZE 就是满，等于 0 就是空——满/空不再歧义，也不用预留空槽。

#define FIFO_SIZE (1u << 4)     // 16，必须是 2 的幂
#define FIFO_MASK (FIFO_SIZE - 1)
// 下标： buf[write_idx & FIFO_MASK]
// 可用： write_idx - read_idx
// 满：   write_idx - read_idx == FIFO_SIZE
// 空：   write_idx - read_idx == 0

二、SPSC 为什么可以无锁

无锁的前提是没有任何一个变量会被双方同时写：

生产者只写 write_idx，只读 read_idx（判满）；
消费者只写 read_idx，只读 write_idx（判空）。

双方各自改自己的字段，不存在"两个上下文同时写一个变量"的竞态，所以不需要互斥锁。这正是 SPSC 与 MPMC（多生产者多消费者）的本质区别：一旦有两方都写同一个索引，就必须加锁或用原子 CAS。

三、内存可见性：别忘了屏障

单核 MCU 上，生产者可能在主循环、消费者在中断里（或反过来）。编译器可能会把索引缓存在寄存器里、或重排"写数据"和"写索引"的顺序，导致消费者读到一个还没写完的槽。解决办法：

索引加 volatile，禁止编译器把它放进寄存器、跨中断优化掉；
在「写数据」和「发布索引」之间加内存屏障，保证数据先落内存、索引后更新。

下面用 GCC 内建全屏障 __sync_synchronize()（ARM 上对应 DMB），简单且正确。

四、完整实现（C 语言，单核/中断友好）

#ifndef SPSC_FIFO_H
#define SPSC_FIFO_H

#include <stdint.h>
#include <stdbool.h>

#define FIFO_SIZE (1u << 4)     /* 16，必须是 2 的幂 */
#define FIFO_MASK (FIFO_SIZE - 1)

typedef struct {
    uint8_t           buf[FIFO_SIZE];
    volatile uint32_t write_idx;   /* 仅生产者写 */
    volatile uint32_t read_idx;    /* 仅消费者写 */
} spsc_fifo_t;

static inline void fifo_init(spsc_fifo_t *f) {
    f->write_idx = 0;
    f->read_idx  = 0;
}

/* 生产者：成功返回 true，满了返回 false */
static inline bool fifo_push(spsc_fifo_t *f, uint8_t byte) {
    uint32_t next = f->write_idx + 1;
    if (next - f->read_idx > FIFO_SIZE)     /* 无符号减法，回绕也正确 */
        return false;
    f->buf[f->write_idx & FIFO_MASK] = byte;
    __sync_synchronize();                   /* 释放屏障：先写数据，后发索引 */
    f->write_idx = next;
    return true;
}

/* 消费者：成功返回 true，空了返回 false */
static inline bool fifo_pop(spsc_fifo_t *f, uint8_t *out) {
    uint32_t w = f->write_idx;
    __sync_synchronize();                   /* 获取屏障：先读索引，后读数据 */
    if (w == f->read_idx)                   /* 空 */
        return false;
    uint8_t b = f->buf[f->read_idx & FIFO_MASK];
    f->read_idx = f->read_idx + 1;
    *out = b;
    return true;
}

/* 当前未消费字节数（生产者/调试用） */
static inline uint32_t fifo_count(spsc_fifo_t *f) {
    return f->write_idx - f->read_idx;
}

#endif

要点：write_idx/read_idx 用 volatile；fifo_push 里先写数据、再用屏障、最后更新索引（发布）；fifo_pop 里先读索引、再用屏障、最后读数据。判满用 next - read_idx > SIZE，判空用 write_idx == read_idx，配合 2^n 下标，没有任何取模、没有空槽浪费。

五、嵌入式中断场景：UART 接收

最典型的用法：串口接收中断当生产者，主循环当消费者。中断里只碰 write_idx，主循环里只碰 read_idx，互不抢。

extern spsc_fifo_t g_uart_fifo;

/* 生产者：UART 接收中断，只碰 write_idx */
void USART1_IRQHandler(void) {
    if (USART1->SR & USART_SR_RXNE) {
        uint8_t c = (uint8_t)USART1->DR;
        fifo_push(&g_uart_fifo, c);
    }
}

/* 消费者：主循环，只碰 read_idx */
int main(void) {
    fifo_init(&g_uart_fifo);
    uart_hw_init();
    uint8_t c;
    while (1) {
        while (fifo_pop(&g_uart_fifo, &c)) {
            app_handle(c);     /* 解析 / 转发 */
        }
        __WFI();               /* 没数据时休眠，等中断唤醒 */
    }
}

相比"中断里直接处理协议"或"关中断拷数据"，这种方式把"收"和"用"彻底解耦：中断只需几条指令把字节塞进 FIFO 就返回，主循环按自己的节奏消费，吞吐和实时性都更好。

六、可运行 demo（PC 端，gcc 直接编）

把第四节的头文件和下面的 main 放一起，gcc fifo_demo.c 即可运行：

#include <stdio.h>
#include "spsc_fifo.h"

int main(void) {
    spsc_fifo_t f;
    fifo_init(&f);

    int dropped = 0;
    for (int i = 0; i < 20; i++) {        /* FIFO 只有 16 字节 */
        if (!fifo_push(&f, (uint8_t)i)) {
            dropped++;
            printf("full, drop %d\n", i);
        }
    }

    uint8_t c;
    int n = 0;
    while (fifo_pop(&f, &c)) {
        printf("%d ", c);
        n++;
    }
    printf("\npopped %d bytes, dropped %d\n", n, dropped);
    return 0;
}

预期输出（容量 16，塞 20 个，后 4 个因满被丢弃）：

full, drop 16
full, drop 17
full, drop 18
full, drop 19
0 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15
popped 16 bytes, dropped 4

从输出可以验证：写进去的 0~15 被原序读出，符合 FIFO；满之后的写入被正确拒绝而不是覆盖未消费数据。

七、多核 SMP 版本（C11 原子，更轻量）

单核 + volatile + 全屏障足够。但到了多核（双核 Cortex-A、Linux 双线程等），不能只靠 volatile——弱内存序下，只用 volatile 会让消费者读到"索引已更新、但数据还没写完"的半成品。正确做法是用 C11 原子，配合 acquire/release 语义，开销比全屏障更小：

#include <stdatomic.h>

typedef struct {
    uint8_t            buf[FIFO_SIZE];
    _Atomic uint32_t   write_idx;
    _Atomic uint32_t   read_idx;
} spsc_fifo_a_t;

static inline bool fifo_push_a(spsc_fifo_a_t *f, uint8_t b) {
    uint32_t w = atomic_load_explicit(&f->write_idx, memory_order_relaxed);
    uint32_t r = atomic_load_explicit(&f->read_idx,  memory_order_acquire);
    if (w + 1 - r > FIFO_SIZE) return false;
    f->buf[w & FIFO_MASK] = b;
    atomic_store_explicit(&f->write_idx, w + 1, memory_order_release);
    return true;
}

static inline bool fifo_pop_a(spsc_fifo_a_t *f, uint8_t *out) {
    uint32_t r = atomic_load_explicit(&f->read_idx,  memory_order_relaxed);
    uint32_t w = atomic_load_explicit(&f->write_idx, memory_order_acquire);
    if (w == r) return false;
    *out = f->buf[r & FIFO_MASK];
    atomic_store_explicit(&f->read_idx, r + 1, memory_order_release);
    return true;
}

这里生产者用 release 发布索引、消费者用 acquire 读取索引，正好配对，保证"看到新索引就一定看到了对应数据"，比 __sync_synchronize() 全屏障更省。

八、坑提醒

必须是严格 SPSC：两个生产者或两个消费者就必须加锁，或改用 MPMC 算法，否则数据错乱。
容量必须是 2^n：& MASK 才能正确闭环；不是 2^n 会下标错位。
索引要用无符号且自由递增：靠 uint32_t 到 42 亿自然回绕，千万不要用取模把索引拉回 0~SIZE-1，否则 write_idx - read_idx 这种差值算法会算错。
单核用 volatile + 屏障；多核用 _Atomic + acquire/release：只靠 volatile 在多核下不安全。
满了别覆盖未消费数据：生产者的正确处理是丢弃 / 覆盖最旧 / 返回错误，绝不能写穿还没被消费的位置。
fifo_push 和 fifo_pop 各只在一个上下文调用：别两边都 push 或都 pop。

小结

SPSC 无锁 FIFO 是嵌入式里性价比极高的一个工具：2^n 容量 + 各改各的索引 + 必要屏障，就能做到无锁、无系统调用、O(1)、Cache 友好。把它用在串口、DMA、传感器采样等"一进一出"的流水线上，能显著降低延迟和 CPU 占用。理解了这套思路，你看任何 RTOS 的 stream buffer、无锁队列，都会觉得"原来就这么回事"。