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 方案一：临界区保护（传统）

#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 方案二：原子操作（无锁）

#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 语义）。修正如下：

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）

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 测试代码（简化）

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