RingBuffer 并发模型——SPSC / ISR / Task

UART ISR 正在往 RingBuffer 写，ProtocolTask 同时在读，为什么它可以不加一把大锁？什么时候又绝对不能这么干？

1. 问题与背景

Producer 通过更新 head，向Consumer发布：新数据已经有效 Read 端也是先读再更新 tail head → Producer发布“数据已产生” tail → Consumer发布“空间已释放”

2. 必要基础

这里还有一个 CPU 层问题

我们现在假设head tail的单次读写不会被撕裂。 现在平台： STM32F407 Cortex-M4 32 bit

而size_t在你的编译环境下通常也是：32 bit 只要自然对齐，普通 32 位 Load / Store 通常就是一条 CPU 指令。

所以不会出现： 先写低16位 ↓ ISR ↓ 再写高16位

这种“半个 head”。 对于我们的 F407，这是一个非常有利的条件。 但这正好说明：无锁不是凭空成立的 如果以后跑到8位 MCU 结果size_t = 16 bit CPU 更新 16 位变量可能要多条指令。 或者某平台：索引宽度 > CPU天然原子访问宽度

那head tail就可能出现 Torn Read / Torn Write。 所以不能说“这个 RingBuffer 在所有 CPU 上天生线程安全。”

准确说法：我们的 SPSC 方案依赖明确的平台原子访问假设；在 STM32F407 这种 32 位 Cortex-M 上，32 位自然对齐索引非常适合这种模型。

这就是为什么架构抽象不等于“不考虑硬件”。

补充：“索引宽度 > CPU 天然原子访问宽度”解释 现在 CPU：Cortex-M4 32 bit CPU 可以粗略理解为它非常擅长一次性读写一个正常对齐的 32 位数据。 例如：uint32_t head; 假设地址正常对齐,CPU读它可能就是一条类似LDR 写：STR(汇编语言)一条指令完成。 所以不会出现： 先读一半 ↓ 被中断 ↓ 再读另一半

但是如果变量变成64位：uint64_t head; CPU 一次天然处理：32 bit 那么一个 64 位变量可能需要：高32位 + 低32位 分多次处理 假设原值 0x00000000_FFFFFFFF 要变成： 0x00000001_00000000

CPU可能类似： 写低32位 ↓ 写高32位

结果刚写完一半， 低位 = 00000000 高位还 = 00000000 此时 ISR 进来读取。

它可能看到0x00000000_00000000 可是这个值从来都不是程序真正想写入的完整新值。

这就叫：Torn Write / Torn Read “撕裂读取 / 撕裂写入”

所以“索引宽度 > CPU天然原子访问宽度”的意思就是 比如：32位 CPU 却让两个执行流共享：64位 index 那么一个 index 的读取/写入：可能无法用一次不可分割的 CPU 访问完成，中断可能夹在中间。

而我们当前： size_t head; size_t tail; 在 F407 编译环境下：size_t = 32 bit Cortex-M4天然32位访问 并且结构体成员正常对齐。 所以这个条件很舒服：32位索引 ≤ CPU自然一次可访问宽度

那么原子到底什么？ 原子操作（Atomic Operation）就是从并发观察者角度看，一个操作不可再分割：别人只能看到“操作之前”或者“操作之后”，不能看到“操作做到一半”。 但是一个非常大的坑：一行 C 代码不代表一个原子操作 “32 位变量在 Cortex-M4 上原子吗？” 对自然对齐的普通 32 位 Load/Store，一般可以作为单次原子访问理解；但像 ++、--、读改写、检查后修改这种复合逻辑并不会因此自动具备原子性。

最开始我理解的是"原子操作对应的是执行一次汇编指令"，但是这不准确， 判断原子性时不应该问：“这一行 C 有几条汇编？” 更应该问：这行代码的修改，需要几个不可被穿插的步骤才能完成？

原子和内存对齐的关系：原子访问通常不仅要求数据宽度 CPU 能天然处理，还要求它满足合适的内存对齐。

uint32_t head; 正常 4 字节对齐：地址 = 0x20000004 Cortex-M4 可以很自然地 一次32位访问，别人不会看到一半。 但如果硬让它放在：0x20000001 这个 32 位数据跨边界。 某些情况下 CPU 需要拆成多个底层访问。 那就可能失去：一次完整不可分割访问这个优势。

“索引宽度 ≤ CPU天然访问宽度 + 正常对齐”

在当前 Cortex-M4 场景中，正常对齐且不超过 CPU 天然访问宽度的简单 Load/Store，通常适合作为原子访问的基础；但原子性的真正定义是“并发观察者不能看到操作的中间状态”，因此它并不严格等于一条汇编指令。

3. volatile

ISR可能修改 head Task正在读取 head

从某个 C 函数自身来看，编译器可能觉得：“这个值在这里没人改啊，我是不是可以把它一直缓存起来？ 但是我们知道ISR可能异步修改它。

所以 volatile 在这里主要告诉编译器：这个变量可能在当前代码看不到的地方发生变化，每次需要它的时候都必须真正进行内存访问，不能随便假设它没变。

注意：volatile ≠ 锁 它不能保证互斥，不能保证多个操作组成原子事务 也不能解决 两个Producer同时修改head

例如： volatile uint32_t count;

TaskA: count++; TaskB: count++; 依然可能 Race。

因为count++ 不是一个抽象上的“原子++”。

通常是： Load ↓ +1 ↓ Store

所以： volatile 解决“编译器可见性/访问行为”问题，不是通用同步机制。

现在要形成一个特别敏感的点：关于变量所有权的问题 head由Producer拥有 tail由Consumer拥有

而reset()函数则会同时修改这两个变量，会侵犯他们的所有权 例子： ISR： 刚写 buffer[5] 准备 head = 6

Task： Reset head = 0 tail = 0

ISR恢复： head = 6

结果就是逻辑状态直接乱 head = 6 tail = 0

所以；RingBuffer_Reset() 只能在通信已经停止、或者外部完成同步后调用。

特别重要的 TOCTOU 思想 例如： if (!RingBuffer_IsEmpty(&rb)) { RingBuffer_ReadByte(&rb, &data); } 先检查 ↓ 后使用

这叫Check Then Act 并发系统里要警惕： 检查时成立 ↓ 真正操作时状态可能改变

现在的单个消费者的前提下比较简单，但是出现多个消费者的情况就很容易出现第二个任务把第一个的数据抢走

例如； TaskA：IsEmpty() → false TaskB：ReadByte() → 把最后一个数据读走 TaskA：ReadByte() 就发现已经空了 真正决定成功与否的，永远应该是 ReadByte() 自己的返回值，而不是前面一次查询。

例如直接： if (RingBuffer_ReadByte(&rb, &data)) { ... } 比： if (!RingBuffer_IsEmpty(&rb)) { RingBuffer_ReadByte(...); }

更稳。 这个思想以后 Queue、文件、网络、资源状态都能用。

4. 局限

当前Producer只可以允许单个生产单个消费，也即是SPSC Multiple Producer 假设：TaskA TaskB 都允许RingBuffer_WriteByte() 当前： head = 3 TaskA：读取 head = 3 next_head = 4 然后被抢占。

TaskB：读取 head = 3 next_head = 4 buffer[3] = 'B' head = 4 TaskA回来： buffer[3] = 'A' head = 4 结果： B 被 A 覆盖

两个 Producer 都认为：自己成功写入了。 但实际只有一个数据存在。 这就是典型：Lost Update / Concurrent Producer Race 多个 Consumer 同理

那我有个问题：为什么对于这种关键变量修改不能套用临界区保护？那会不会伴随架构的问题呢？ 不要为了让底层 RingBuffer自己支持： 10 Producer 5 Consumer

就把它内部搞成： Mutex Semaphore Critical Section FreeRTOS Scheduler 那 RingBuffer 会越来越胖。

更好的做法： 底层工具 → 保持简单、明确能力边界 上层架构 → 把复杂并发整理成工具能处理的模型

不是让底层适应所有混乱，而是通过架构让上层产生更干净的数据流。 以及另一个原因： 高频通信数据每个字节都关一段可屏蔽中断： 实时性也受影响

所以第一选择不是“看到并发就 Critical Section。” 而是 先设计 Ownership，让冲突尽量不存在。

锁应该解决：确实无法通过Ownership消除的共享访问，而不是不是修补所有架构问题。

此阶段要形成的条件映射 以后看到共享资源，不应该第一句想“这里要不要加 Mutex？”

先想 有几个Writer？ 有几个Reader？ 谁拥有它？ 能不能通过架构把多个Writer归并成一个Owner？ ISR会不会访问？ 数据发布顺序是什么？

然后才决定： SPSC Queue Mutex Critical Section Single Owner Task 这比“并发 = 加锁”高一个层次。

第一，SPSC RingBuffer 可以通过严格的数据所有权降低同步复杂度：Producer 只推进 head，Consumer 只推进 tail。

第二，Producer 应该先写入数据，最后更新 head 发布数据；Consumer 先读数据，再推进 tail 释放空间。

第三，volatile 可以用于表达 ISR 等异步执行环境可能修改索引，但它本身不是锁，也不能解决多个执行流同时修改同一状态的问题。

第四，多 Producer / 多 Consumer 会破坏这个简单模型，需要外部同步或通过 Queue + Single Owner 将数据流重新整理为 SPSC。

第五，是否真正能够无锁还依赖平台对索引访问的原子性等假设，因此不能脱离 CPU 和编译器环境笼统宣称“线程安全”。

5. RingBuffer 批量读写 + 环形内存的连续性问题

size_t RingBuffer_Write(RingBuffer_t *rb,const uint8_t *data,size_t length) { size_t free_size; size_t write_size; size_t first_size; size_t second_size;

if ((rb == NULL) ||(rb->buffer == NULL) ||(data == NULL) ||(length == 0U))
{
    return 0U;
}

free_size = RingBuffer_Free(rb);

write_size =(length < free_size) ?length :free_size;

if (write_size == 0U)
{
    return 0U;
}

first_size = rb->capacity - rb->head;

if (first_size > write_size)
{
    first_size = write_size;
}

memcpy(&rb->buffer[rb->head],data,first_size);
second_size = write_size - first_size;

if (second_size > 0U)
{
    memcpy(&rb->buffer[0],&data[first_size],second_size);
}

rb->head =(rb->head + write_size) % rb->capacity;

return write_size;

}

size_t RingBuffer_Read(RingBuffer_t *rb,uint8_t *data,size_t length) { size_t available; size_t read_size; size_t first_size; size_t second_size;

if ((rb == NULL) ||(rb->buffer == NULL) ||(data == NULL) ||(length == 0U))
{
    return 0U;
}

available = RingBuffer_Available(rb);

read_size = (length < available) ?length :available;

if (read_size == 0U)
{
    return 0U;
}

first_size = rb->capacity - rb->tail;

if (first_size > read_size)
{
    first_size = read_size;
}

memcpy(data,&rb->buffer[rb->tail],first_size);

second_size = read_size - first_size;

if (second_size > 0U)
{
    memcpy(&data[first_size],&rb->buffer[0],second_size);
}
rb->tail =(rb->tail + read_size) %rb->capacity;

return read_size;

}

但注意：多次copy大型数组也会影响实时性

那为什么不能让 DMA 直接写 RingBuffer？ 因为 DMA 非常喜欢： 地址 + 长度 + 一整段连续空间

例如：请从 buffer[100]连续写 80 Bytes 但 RingBuffer 当前：head = 220 capacity = 256 物理上是尾部：220 ~ 255 = 36 Bytes 开头：0 ~ 63 = 64 Bytes

普通 DMA 一次传输通常不能把：前36 Byte写尾部 自动跳回 再写64 Byte 这就是RingBuffer 和 DMA 第一次真正发生矛盾的地方

我们后面会出现两条架构路线 路线 A：DMA Staging Buffer + Copy UART ↓ DMA ↓ 固定连续 DMA Buffer ↓ ISR / Callback ↓ RingBuffer_Write() ↓ ProtocolTask 优点：简单 可靠 DMA配置容易 RingBuffer独立 缺点：多一次 memcpy，大型数组会影响实时性

这是很多普通 MCU 项目完全可以接受的方案。

路线 B：Zero Copy / Direct Buffer Access 想办法让DMA直接拿到RingBuffer当前连续可写区域

例如 RingBuffer 告诉 DMA： 当前从 head 开始 有 36 Bytes 连续空间 地址是 &buffer[220]

DMA直接写，完成以后Commit 36 Bytes RingBuffer 才更新 head。 这就变成： Reserve ↓ DMA Write ↓ Commit

这里已经非常接近真正高性能通信框架了，但是相对的代码复杂度也就上升了 必须要解决 这块内存现在属于谁？ DMA写完了吗？ Task能不能提前读？ 什么时候更新head？ Wrap以后第二段怎么办？ DMA Error以后这块空间怎么处理？ Cache平台怎么办？

所以Zero Copy 不是天然比 Copy 高级。 对于 STM32F407 这种项目：数据量不大 CPU负载有余量 一两次 memcpy 往往比复杂 Ownership Bug 划算得多。

这也是工程思维：性能优化要有成本意识。

不要一看到memcpy就觉得：“性能低，我一定要 Zero Copy。” 应该问： 数据吞吐量到底多少？ CPU利用率多少？ 内存够吗？ Latency要求多高？ Copy是不是实际瓶颈？

如果不是瓶颈：简单方案优先 因为： 简单= 更容易验证 更容易维护 更少并发Bug

6. RingBuffer vs FreeRTOS StreamBuffer

RingBuffer 主要解决“数据怎么存” StreamBuffer 在此基础上进一步解决“Task 怎么等数据、ISR 怎么唤醒 Task、生产者和消费者怎么通过 RTOS 调度协作”。

StreamBuffer = 环形字节缓冲区 + RTOS等待/唤醒能力

不同 FreeRTOS 版本字段细节可能有变化，但概念上你可以想成： StreamBuffer ├─ Storage Buffer ├─ Length ├─ Head ├─ Tail ├─ Trigger Level ├─ Waiting Reader └─ Waiting Writer

只是 FreeRTOS 又加了： 哪个Task正在等数据 哪个Task正在等空间 什么时候应该唤醒

这就是它从Data Structure升级成RTOS IPC Object

Trigger Level 是什么？ 等待接收的 Task 希望在满足相应条件时，至少积累到一定数据量再被唤醒，而不是每来 1 Byte 都急着叫醒。

RingBuffer = 纯数据结构，解决“数据怎么存”

StreamBuffer = 字节流Buffer + Task Block + Wakeup + Timeout + FromISR

裸机 / 跨平台工具 → RingBuffer更有优势 FreeRTOS内部Task通信 → StreamBuffer非常方便

目前至少有两套成熟路线： 方案 A DMA / ISR ↓ RingBuffer ↓ Task Notification ↓ CommTask

方案 B DMA / ISR ↓ StreamBuffer ↓ CommTask

具体选哪一个，应该是比较： 吞吐量 Copy次数 ISR执行时间 Buffer Ownership DMA连续内存 可移植性 协议解析模型

7. MessageBuffer —— “字节流”和“消息”到底有什么区别

StreamBuffer 传的是“连续字节流”，MessageBuffer 传的是“一条一条完整消息”。 假设发送两次：第一次发送ABC，第二次发送DEF StreamBuffer质上只看到：A B C D E F 它不关心： ABC 是一条消息 DEF 是另一条消息

对于接收者来说，这只是Byte Stream

MessageBuffer 会保留这个边界：[ ABC ] [ DEF ] 接收方一次 Receive拿到 ABC，下一次拿到 DEF

不会把它们糊成：ABCDEF

所以MessageBuffer 的核心价值不是“能存字节”，而是“保留消息边界”。

通信系统中非常重要的三层概念一定要分清： Byte ↓ Frame ↓ Message

例如 UART 收到AA 55 01 03 10 20 30 7F 其中：AA 55 → 帧头 01→ Command 03→ Payload Length 10 20 30→ Payload 7F→ CRC

这一整坨AA 55 01 03 10 20 30 7F 叫Frame，它是在通信协议层定义的完整数据单元。 而解析之后，我们可能得到： typedef struct { uint8_t cmd; uint8_t payload[32]; size_t length; } ProtocolMessage_t;

这已经不是“原始 UART 字节”了，而是业务消息 比如cmd = MOTOR_SET_SPEED payload = 1000 RPM

更好的数据流： UART Hardware ↓ Raw Bytes ↓ Protocol Parser ↓ Complete Frame ↓ Message ↓ Service / App 可以把它理解成： 物理世界 ↓ 字节 ↓ 协议 ↓ 语义

MessageBuffer 和 Queue 又有什么区别？ Queue 更偏固定大小对象；MessageBuffer 更偏可变长度消息。

机制 数据语义 RingBuffer 通用环形数据结构 Queue 一个个固定大小对象 StreamBuffer 连续字节流 MessageBuffer 一条条可变长消息

RingBuffer：不负责 Block/Wakeup 后三个：都是 RTOS 同步/通信机制

8. 真实工程中的用途

UART RX 应该怎么设计？ 第一层： UART DMA ↓ Raw Bytes 只是Raw Byte Stream 这段是原始字节RingBuffer + Notification或者StreamBuffer都合理。

第二层： Raw Bytes ↓ Protocol Parser

Parser负责： 找帧头 读Length 收Payload CRC校验 完整帧判断 超时恢复 错帧恢复

ProtocolTask解析之后： 如果得到 固定大小 AppEvent 用Queue 如果得到 大量可变长度 Message 考虑MessageBuffer 注意一个点：接收 Buffer 必须足够容纳完整下一条消息，否则不能把消息拆开交给你。

第三层： Complete Frame ↓ Decode ↓ Message

第四层： Message ↓ Service / App

此时才进入： MotorService SensorService SystemManager

StreamBuffer 传输的是连续字节流，不保留每次发送之间的边界；MessageBuffer 会保留消息边界。 Queue 更适合固定大小对象，MessageBuffer 更适合可变长度完整消息。 UART 原始接收数据天然属于 Byte Stream，应该先经过 Protocol Parser，再转换成 Frame / Message，不能让 APP 直接解析原始字节。 IPC 的选择应该由“数据语义 + Ownership + 并发模型”决定，而不是哪个 FreeRTOS API 看起来更高级。

9. STM32F407 UART + DMA + IDLE 接收骨架

RX数据通路： UART ↓ DMA ↓ DMA 临时缓冲区 ↓ IDLE / DMA完成中断 ↓ Platform 层 ↓ SPSC RingBuffer ↓ Task Notification ↓ CommTask ↓ 协议层（下一章）

最简单写法： static uint8_t rx_dma_buf[128]; DMA 收完： Callback ↓ 拷进 RingBuffer ↓ 重新开启 DMA

问题是DMA 没重新打开之前，UART 又来了数据怎么办？有一个接收空窗期。 而如果我们为了减少空窗，先重新开启 DMA： 重新让DMA使用 rx_dma_buf ↓ 然后才 memcpy

又出现：DMA正在写 + CPU正在读 同一个 Buffer Ownership 冲突。

所以第一版工程骨架我们直接用：

Ping-Pong DMA Buffer Buffer A Buffer B

例如： 当前 DMA → A A完成 ↓ 立刻切 DMA → B ↓ CPU处理A

下一轮： DMA → A CPU处理B

这就是前面一直要学的Ownership