# ESP32 双核环境下用原子操作替代临界区保护共享 FIFO 的实战对比 ## 1. 引言 在 ESP32 双核 FreeRTOS 应用中,多个任务(如传感器采集、网络发送)常共享一个 FIFO 缓冲区。传统做法是使用临界区(`taskENTER_CRITICAL`/`taskEXIT_CRITICAL`)保护读写操作,但临界区会关闭中断或屏蔽调度器,导致高优先级任务和中断响应延迟。ESP32 提供了硬件原子操作(如 `portMUX_TYPE` 和 `atomic` 指令),可以在不阻塞中断的情况下实现无锁并发,显著提升实时性。本文通过一个实际 FIFO 示例,对比两种方案的性能与可靠性。 ## 2. 原理分析 ### 2.1 临界区(Critical Section) - 在单核 MCU 上,临界区通常通过关中断实现;在 ESP32 双核上,FreeRTOS 使用 `portMUX_TYPE` 自旋锁,防止多核同时进入临界区。 - 缺点:进入临界区会阻塞其他核心和中断,若临界区代码较长,会严重影响系统实时性。 ### 2.2 原子操作(Atomic Operations) - ESP32 基于 Xtensa LX6 双核,支持 `atomic` 指令(如 `S32C1I`),FreeRTOS 提供 `portMUX_TYPE` 和 `atomic` 函数(如 `portENTER_CRITICAL` 和 `atomicCAS`)。 - 原子操作在硬件层面保证读-改-写操作的不可分割性,无需关闭中断,适合保护短小的共享变量(如 FIFO 的读写索引)。 - 对于 FIFO,我们可以使用原子操作更新读/写指针,并结合内存屏障保证数据可见性。 ## 3. 实战对比:共享 FIFO 实现 ### 3.1 硬件环境 - ESP32 DevKitC,双核 240MHz,FreeRTOS 10.4.3。 ### 3.2 需求 - 一个环形 FIFO,容量 256 字节,一个生产者任务(core 0)写入,一个消费者任务(core 1)读取。 - 要求:数据不丢失,且写入/读取延迟尽量低。 ### 3.3 方案一:临界区保护(传统) ```c #include "freertos/FreeRTOS.h" #include "freertos/task.h" #include "esp_attr.h" #define FIFO_SIZE 256 typedef struct { uint8_t buffer[FIFO_SIZE]; volatile uint32_t head; // 写入位置 volatile uint32_t tail; // 读取位置 portMUX_TYPE mux; } fifo_t; void fifo_init(fifo_t *f) { f->head = f->tail = 0; f->mux = portMUX_INITIALIZER_UNLOCKED; } bool fifo_write(fifo_t *f, uint8_t data) { portENTER_CRITICAL(&f->mux); uint32_t next = (f->head + 1) % FIFO_SIZE; if (next == f->tail) { portEXIT_CRITICAL(&f->mux); return false; // 满 } f->buffer[f->head] = data; f->head = next; portEXIT_CRITICAL(&f->mux); return true; } bool fifo_read(fifo_t *f, uint8_t *data) { portENTER_CRITICAL(&f->mux); if (f->head == f->tail) { portEXIT_CRITICAL(&f->mux); return false; // 空 } *data = f->buffer[f->tail]; f->tail = (f->tail + 1) % FIFO_SIZE; portEXIT_CRITICAL(&f->mux); return true; } ``` ### 3.4 方案二:原子操作(无锁) ```c #include "freertos/FreeRTOS.h" #include "freertos/task.h" #include "esp_attr.h" #include "esp_atomic.h" typedef struct { uint8_t buffer[FIFO_SIZE]; volatile uint32_t head; // 写入位置,原子更新 volatile uint32_t tail; // 读取位置,原子更新 } fifo_atomic_t; void fifo_atomic_init(fifo_atomic_t *f) { f->head = f->tail = 0; } bool fifo_atomic_write(fifo_atomic_t *f, uint8_t data) { uint32_t head, next; do { head = atomic_load(&f->head); next = (head + 1) % FIFO_SIZE; if (next == atomic_load(&f->tail)) { return false; // 满 } } while (!atomic_compare_exchange_weak(&f->head, &head, next)); // 写入数据(此时 head 已更新,但数据写入需在 head 更新前完成?注意顺序) f->buffer[head] = data; // 内存屏障确保数据写入对其他核心可见 atomic_thread_fence(memory_order_release); return true; } bool fifo_atomic_read(fifo_atomic_t *f, uint8_t *data) { uint32_t tail, next; do { tail = atomic_load(&f->tail); if (tail == atomic_load(&f->head)) { return false; // 空 } next = (tail + 1) % FIFO_SIZE; } while (!atomic_compare_exchange_weak(&f->tail, &tail, next)); // 读取数据 *data = f->buffer[tail]; atomic_thread_fence(memory_order_acquire); return true; } ``` **注意**:上述原子版本存在一个潜在问题:写入数据时,head 已更新,但数据尚未写入缓冲区,消费者可能读到旧数据。正确做法是:先写入数据,再更新 head(使用 release 语义)。修正如下: ```c bool fifo_atomic_write(fifo_atomic_t *f, uint8_t data) { uint32_t head, next; do { head = atomic_load(&f->head); next = (head + 1) % FIFO_SIZE; if (next == atomic_load(&f->tail)) { return false; } // 先写入数据 f->buffer[head] = data; // 内存屏障确保数据写入对其他核心可见 atomic_thread_fence(memory_order_release); // 尝试更新 head } while (!atomic_compare_exchange_weak(&f->head, &head, next)); return true; } bool fifo_atomic_read(fifo_atomic_t *f, uint8_t *data) { uint32_t tail, next; do { tail = atomic_load(&f->tail); if (tail == atomic_load(&f->head)) { return false; } next = (tail + 1) % FIFO_SIZE; // 读取数据 *data = f->buffer[tail]; atomic_thread_fence(memory_order_acquire); // 尝试更新 tail } while (!atomic_compare_exchange_weak(&f->tail, &tail, next)); return true; } ``` 但这里存在一个问题:写入时,如果 CAS 失败(因为其他核心修改了 head),则数据已写入 buffer[head],但 head 未更新,导致数据被覆盖或丢失。因此,这种写法仅适用于单生产者单消费者(SPSC)场景,且 head 和 tail 的更新顺序需要保证。对于 SPSC,可以使用更简单的方案:head 和 tail 各自独立,且只有一个生产者一个消费者,无需 CAS,只需原子递增。但 ESP32 双核上,原子递增也是必要的。 为了文章简洁,我们采用 SPSC 场景,并使用原子递增(`atomic_fetch_add`)实现无锁 FIFO。 ### 3.5 改进的无锁 FIFO(SPSC) ```c typedef struct { uint8_t buffer[FIFO_SIZE]; volatile uint32_t head; // 生产者写索引 volatile uint32_t tail; // 消费者读索引 } fifo_spsc_t; void fifo_spsc_init(fifo_spsc_t *f) { f->head = f->tail = 0; } bool fifo_spsc_write(fifo_spsc_t *f, uint8_t data) { uint32_t head = atomic_load(&f->head); uint32_t next = (head + 1) % FIFO_SIZE; if (next == atomic_load(&f->tail)) { return false; } f->buffer[head] = data; atomic_store(&f->head, next); // 释放语义,确保数据先写入 return true; } bool fifo_spsc_read(fifo_spsc_t *f, uint8_t *data) { uint32_t tail = atomic_load(&f->tail); if (tail == atomic_load(&f->head)) { return false; } *data = f->buffer[tail]; atomic_store(&f->tail, (tail + 1) % FIFO_SIZE); // 释放语义 return true; } ``` 注意:`atomic_store` 默认是顺序一致,但我们可以使用 `atomic_store_explicit` 指定 memory_order_release。在 ESP32 上,`atomic_store` 通常足够。 ## 4. 性能对比测试 ### 4.1 测试方法 - 创建两个任务,分别运行在 core 0 和 core 1。 - 生产者任务循环写入 10000 次,消费者任务循环读取。 - 使用 `esp_timer` 测量总耗时,并统计平均每次操作时间。 - 分别运行临界区版本和原子操作版本。 ### 4.2 测试代码(简化) ```c void producer_task(void *arg) { fifo_t *f = (fifo_t*)arg; for (int i = 0; i < 10000; i++) { while (!fifo_write(f, (uint8_t)i)); } vTaskDelete(NULL); } void consumer_task(void *arg) { fifo_t *f = (fifo_t*)arg; uint8_t data; for (int i = 0; i < 10000; i++) { while (!fifo_read(f, &data)); } vTaskDelete(NULL); } ``` ### 4.3 结果对比(示例数据) | 方案 | 平均写入时间 (us) | 平均读取时间 (us) | 总耗时 (ms) | |------|------------------|------------------|-------------| | 临界区 | 12.5 | 11.8 | 125 | | 原子操作 | 3.2 | 2.9 | 32 | (实际数据因环境而异,但原子操作通常快 3-4 倍) ## 5. 注意事项 - **原子操作适用场景**:仅适用于保护简单的共享变量(如索引、标志),对于复杂数据结构(如链表)仍需锁。 - **内存顺序**:务必使用正确的内存屏障(`memory_order_release`/`acquire`)确保数据可见性。 - **SPSC vs MPSC**:上述无锁 FIFO 仅适用于单生产者单消费者;多生产者多消费者需要更复杂的算法(如使用 CAS 循环)。 - **临界区并非不可用**:对于短临界区,临界区开销很小,且更易实现;对于长临界区,考虑使用互斥量或队列。 - **调试难度**:无锁编程调试困难,建议先用临界区实现,再优化。 ## 6. 总结 在 ESP32 双核环境下,使用原子操作替代临界区保护共享 FIFO 可以显著降低延迟,提高吞吐量,但需要仔细设计算法并理解内存模型。对于简单场景(如 SPSC),原子操作是理想选择;对于复杂场景,临界区或互斥量更安全。开发者应根据实际需求权衡。 ## 7. 参考资料 - ESP-IDF Programming Guide: Atomic Operations - FreeRTOS Documentation: Critical Sections and Atomic Operations