如何用原子操作正确同步多读单写线程及双缓冲切换？

我正在开发一套数据处理系统，该系统持续接收信息流并缓冲，偶尔触发数据处理。为此采用双缓冲方案：一个缓冲存储输入信息，另一个缓冲进行处理，相关代码如下：

struct data_bundle_1 { /* ... */ };
struct data_bundle_2 { /* ... */ };
// ...
struct data_bundle_n { /* ... */ };

struct data_buffer {
    data_bundle_1 data_1;
    data_bundle_2 data_2;
    // ...
    data_bundle_n data_n;
    std::atomic_int buffer_writer_count;
};

data_buffer double_buf[2];
std::atomic<data_buffer*> current_buf;
int current_buf_index = 0;

数据缓冲由多个写线程填充，因此data_buffer中的每个data_bundle都带有内部锁（大概率是基于atomic_flag的自旋锁）以避免数据竞争。我不确定如何在切换缓冲时避免数据竞争：操作current_buf时，写线程变为读线程，缓冲切换线程是唯一写线程。我提出了一个实现方案，但不确定是否正确：

// 线程#1,2,...,m
void data_writer_routine (/* data batch */) {
    data_buffer* cur_buf = current_buf.load(std::memory_order_acquire); // (1)
    cur_buf->buffer_writer_count.fetch_add(1, std::memory_order_release); // (2)
    {
        // 防止(1)和(2)之间发生缓冲切换
        data_buffer* new_cur_buf = current_buf.load(std::memory_order_acquire);
        if (cur_buf != new_cur_buf) {
            cur_buf->buffer_writer_count.fetch_sub(1, std::memory_order_release);
            (cur_buf = new_cur_buf)->buffer_writer_count.fetch_add(1, std::memory_order_release);
        }
    }
    /* 将数据批次加入缓冲 */
    cur_buf->buffer_writer_count.fetch_sub(1, std::memory_order_release); // (3)
}

// 线程#m+1
void buffer_switching_routine () {
    data_buffer* old_buf = current_buf.exchange(
        &(double_buf[(current_buf_index=1-current_buf_index)]),
        std::memory_order_release); // (4)
    // 等待所有写线程完成对旧缓冲的操作
    while (old_buf->buffer_writer_count.load(std::memory_order_acquire)); // (5)
    /* 处理old_buf中的数据 */
}

我需要明确三个问题：

1. 线程#m+1执行语句(4)后，线程#1-m后续执行语句(1)时，是否能保证获取到(4)设置的新值？
2. 我对内存序常量的使用是否正确？比如同一线程对同一原子变量多次执行acquire操作是否可行？跨线程中单个release对应多个acquire、单个acquire对应多个release是否可行？
3. 如果上述实现存在问题（即无法保证问题1的要求，线程#1-m可能获取到旧缓冲指针引发数据竞争），仅使用原子操作的正确同步方式是什么？我了解std::shared_mutex，但希望避免内核级同步带来的性能损耗。

问题解答 #

让我逐个帮你拆解这些问题，结合C++内存模型和双缓冲的实践来分析：

问题1：切换后写线程能否保证读到新缓冲指针？ #

完全可以保证。

根据C++内存模型，切换线程执行的current_buf.exchange(..., std::memory_order_release)是一个release操作，而写线程执行的current_buf.load(std::memory_order_acquire)是acquire操作。当一个release操作完成后，所有后续的acquire操作读取同一个原子变量时，会与这个release操作建立同步关系（synchronizes-with）——这意味着切换线程对current_buf的修改，对所有后续执行acquire加载的写线程是完全可见的。

换句话说，一旦切换线程执行完语句(4)，任何写线程之后执行语句(1)时，必然会读取到新的缓冲指针，不会再拿到旧值。

问题2：内存序的使用是否正确？ #

整体思路是对的，但部分内存序可以优化，同时你的疑问点都是合法的：

• 同一线程多次执行acquire操作完全可行：内存序规则约束的是跨线程的内存可见性，同一线程内的原子操作遵循as-if规则，执行顺序是确定的，多次acquire不会有任何问题。
• 单个release对应多个acquire是合法的：一个release操作可以和任意多个后续的acquire操作建立同步关系，所有这些acquire都能看到release之前的所有修改。
• 单个acquire对应多个release：如果多个release操作修改同一个原子变量，后续的acquire会看到最后一次release的结果，这也是符合内存模型的（不过这种场景在你的代码里不存在）。

针对你代码中的内存序细节优化建议：

• 语句(2)和双重检查里的fetch_add/fetch_sub用std::memory_order_relaxed即可，不需要release。因为这些计数操作的同步依赖于current_buf的acquire/release，我们只需要保证计数的原子性，不需要额外的内存可见性约束。
• 语句(3)的fetch_sub保留release是对的，它需要让切换线程的load(acquire)（语句5）能及时看到计数的递减，确保等待逻辑正确。

问题3：纯原子操作的优化实现 #

你的原始实现逻辑是正确的，但双重检查的写法可以更简洁健壮，同时可以去掉冗余的current_buf_index变量。这里提供一个优化后的版本：

优化后的写线程代码

void data_writer_routine(/* data batch */) {
    data_buffer* cur_buf;
    do {
        // 先加载当前缓冲指针
        cur_buf = current_buf.load(std::memory_order_acquire);
        // 递增计数（仅保证原子性，用relaxed足够）
        cur_buf->buffer_writer_count.fetch_add(1, std::memory_order_relaxed);
        // 再次检查缓冲是否切换，防止fetch_add前发生了切换
        if (cur_buf == current_buf.load(std::memory_order_acquire)) {
            break;
        }
        // 缓冲已切换，回滚旧缓冲的计数
        cur_buf->buffer_writer_count.fetch_sub(1, std::memory_order_relaxed);
    } while (true);

    /* 将数据批次加入缓冲 */

    // 递减计数，用release保证切换线程能看到这个修改
    cur_buf->buffer_writer_count.fetch_sub(1, std::memory_order_release);
}

这个循环确保写线程最终拿到的是当前有效的缓冲指针，且计数正确递增，避免了原始代码中单次检查可能存在的竞态。

优化后的切换线程代码

void buffer_switching_routine() {
    // 直接通过当前缓冲指针计算另一个缓冲，无需全局index变量
    data_buffer* new_buf = &double_buf[current_buf.load(std::memory_order_relaxed) == &double_buf[0] ? 1 : 0];
    // 切换缓冲，release保证所有后续acquire能看到新指针
    data_buffer* old_buf = current_buf.exchange(new_buf, std::memory_order_release);
    // 自旋等待所有写线程离开旧缓冲，acquire保证能看到计数的递减
    while (old_buf->buffer_writer_count.load(std::memory_order_acquire)) {
        // 加入yield减少CPU空转消耗
        std::this_thread::yield();
    }
    /* 处理old_buf中的数据 */
}

去掉了current_buf_index，通过指针直接判断要切换到的缓冲，逻辑更简洁且避免了额外的全局变量竞争。

内容的提问来源于stack exchange，提问作者Semisonic