C++ / a working model

74 / 80   ·   C++11   ·   约 10 分钟

原子操作:data race、RMW 与无锁边界

先记住这句话

atomic 解决单个原子对象访问不可分割的问题,不会自动把多步业务变成事务。计数更新需要 fetch_add 等读改写操作;原子类型也不承诺一定无锁,更不保证任何基于它的算法都无等待。

本篇内容
  1. 数据竞争不是偶尔算错
  2. 读改写必须是同一个原子操作
  3. 原子性、无锁与性能是三件事
  4. 运行示例
  5. 动手练习

数据竞争不是偶尔算错

不同线程对同一内存位置发生冲突访问,其中至少一个不是原子操作,又没有先行发生关系把它们排序,就可能构成 data race,导致未定义行为。冲突不仅包括一读一写,也包括两个写入;即使写入相同数值,也不能因此认为安全。普通计数器的并发自增不是可接受的近似统计。

把计数器换成 std::atomic<int>,并只通过原子接口操作,就能避免这个对象上的数据竞争。但业务上的 race condition 范围更广:程序可以没有数据竞争,却仍因“检查再操作”被别的线程插入而违反逻辑。判断线程安全时必须同时检查内存合法性和业务不变量。

读改写必须是同一个原子操作

load 只读,store 只写,fetch_add 是不可分割的 read-modify-write,返回修改前的值。counter.store(counter.load() + 1) 是两个操作;两个线程都可能先读到零,再分别写入一,合法却丢掉一次增量。fetch_add(1) 或原子自增才表达需要的整体更新。

示例让工作线程与主线程各增加一千次,最后 join 后读取两千。这里计数不负责发布其他数据,所以更新可用 relaxed;join 提供任务结束的同步边界。复杂条件更新可用 compare_exchange:比较成功才提交,失败时 expected 会被实际观察值更新。weak 版本允许虚假失败,通常放在重试循环里,循环中应重新计算依赖旧值的目标值。

原子性、无锁与性能是三件事

标准允许许多 atomic<T> 的实现内部使用锁。is_lock_free() 查询对象上的操作是否无锁,C++17 的 is_always_lock_free 是类型层面的编译期保证;不能从名字或对象大小推断。atomic_flag 的原子操作保证无锁,但围绕它写出的自旋锁依然可能等待另一个线程释放。

无锁也不等于每个线程都在有限步骤内完成,更不等于快。高竞争的同一缓存行会造成通信成本;把多个字段分别改成 atomic 也不能得到一致快照。复杂共享结构优先用互斥锁保护整体状态,确有需求再设计原子协议。volatile 不提供线程间原子性或同步关系,不能代替这些工具。

容易答错的地方

  • 原子对象默认的 seq_cst 顺序也不能把分开的 load 与 store 合并成一次原子自增;更强内存序不修复操作粒度。
  • 原子指针只保护指针值,不自动保护所指对象的读写,也不会阻止另一个线程提前释放该对象。

运行一个例子

最低标准 C++11 · 完整程序 · 下载 .cpp

#include <atomic>
#include <cassert>
#include <iostream>
#include <thread>

int main() {
    std::atomic<int> count{0};
    const auto increment = [&] {
        for (int i = 0; i < 1000; ++i) {
            count.fetch_add(1, std::memory_order_relaxed);
        }
    };
    std::thread worker(increment);
    increment();
    worker.join();
    const int total = count.load(std::memory_order_relaxed);
    assert(total == 2000);

    const int old = count.fetch_add(5, std::memory_order_relaxed);
    assert(old == 2000);
    assert(count.load(std::memory_order_relaxed) == 2005);
    std::cout << total << ' ' << old + 5 << '\n';
}

在本地编译

g++ -std=c++11 -Wall -Wextra -Wpedantic -pthread concurrency-atomics.cpp -o example && ./example

预期结果

2000 2005

CHECK YOUR UNDERSTANDING

合上答案,试着解释。

两个线程用 if (tickets.load() > 0) tickets.fetch_sub(1) 领取最后一张票。全用原子操作就能保证票数不为负吗?给出修复思路。

查看参考答案

不能。两线程都可能先看到一,再分别扣减,得到零和负一。应把条件与扣减合并:令 expected = tickets.load();循环条件为 expected > 0;尝试 compare_exchange_weak(expected, expected - 1),成功就领取并退出,失败用被更新的 expected 重新判断。计数本身不发布其他状态时可使用 relaxed;也可以直接用一把 mutex 保护检查和扣减。

继续查证

标准草案链接会随工作草案更新;本文版本标记对应示例最低要求,不表示草案中的所有新规则都适用于旧标准。

回到目录