从 AZULC 中学 Rust 02：理解原子操作

上一章让下载任务和上报任务共同拥有 ProgressTracker。对象不会提前释放了，但还有一个问题：几个任务怎样往同一个“已下载字节数”里加数据？

源码入口：ProgressTracker · 游戏就绪标志 · Worker 计数

1 从普通数字到“原子数字”

一个任务计数很简单：

fn main() {
    let mut downloaded: u64 = 0;
    downloaded += 100;
    println!("{downloaded}");
}

当多个独立任务出现时，却不能同时持有同一个普通数字的独占可变引用。Rust 限制这种访问，避免发生未受保护的并发读写。

如果共享的只是一个计数，可以使用 AtomicU64。U64 表示保存无符号 64 位整数；Atomic 表示提供原子操作，此时的一次操作不会被另一个任务拆开插入。

use std::sync::atomic::{AtomicU64, Ordering};

fn main() {
    let downloaded = AtomicU64::new(0);
    let current = downloaded.load(Ordering::Relaxed);
    assert_eq!(current, 0);
}

我们将在后面章节详细解析这段代码。

2 原子操作方法

2.1 load：读取，而不做修改

AZULC 的界面定期读取下载或安装进度，就是在使用 load。源码的 snapshot 方法里有：

// 源码节选
current: self.current.load(Ordering::Relaxed),

读出来的是当次观察到的数字。随后别的任务继续累加，这个普通整数不会自动变化；界面下次更新需要重新 load。

因此 load 返回的是“这次观察到多少”，不是给你一个会跟着共享状态自动更新的变量。

2.2 store：设为指定值，而不返回旧值

假设现有一个原子数字，原来是 100，现在要设为 250：

use std::sync::atomic::{AtomicU64, Ordering};

fn main() {
    let downloaded = AtomicU64::new(100);
    downloaded.store(250, Ordering::Relaxed);
    assert_eq!(downloaded.load(Ordering::Relaxed), 250);
}

注意到这里 downloaded 没有声明 mut，也能调用 store，这是因为原子类型提供了通过共享引用进行受控修改的接口，即内部可变性，而非把里面的数值变成可以任意操作的普通可变引用。

对于 AZULC 登录账号时的状态，我们维护一个布尔开关 AtomicBool 。取消登录时把它设为 true，检查方 load 观察。如果还需要传递取消原因，就不仅仅是一个开关的问题了，可以考虑第 6 篇的消息通道。

2.3 load 可以直接叠加使用 store 吗？

假设现有一个原子数字，不知道其原始值，现在的任务是要给它加 100：

let old = downloaded.load(Ordering::Relaxed);
downloaded.store(old + 100, Ordering::Relaxed);

一个任务独自运行时这没问题，但如果有两个任务，这两步却可能这样交错：

两个任务各下载了 100，最后只有 100。没有哪次原子访问被拆坏；问题是读取与写入分成了两次调用，中间允许另一个任务进入。这就是逻辑竞争。

2.4 fetch_add：把读取、相加、写回合成一次

在 AZULC 的下载进度跟踪器 ProgressTracker 中，在收到一个数据块后会执行：

// 源码节选：count 是本次数据块的字节数。
tracker.current.fetch_add(count, Ordering::Relaxed);

fetch_add 完成一次不可分割的“取旧值、加上参数、保存”，返回的是旧值：

use std::sync::atomic::{AtomicU64, Ordering};

fn main() {
    let downloaded = AtomicU64::new(100);
    let previous = downloaded.fetch_add(50, Ordering::Relaxed);
    assert_eq!(previous, 100);
    assert_eq!(downloaded.load(Ordering::Relaxed), 150);
}

现在两个任务累加：

B 也可以先执行，但都能累计到 200。不会因为同时拿到旧值而覆盖别人的结果。

同类的 fetch_sub 用于减法：项目重试时回退本轮已统计的字节，Worker 结束时减少工作计数。业务仍要保证只减自己的部分。原子整数加减可以回绕，也不会替你验证“占用不能减成负数”这样的业务约束。

2.5 swap：保存新值，同时知道此前是什么

游戏启动日志可能多次出现就绪特征，但 UI 只需要接收一次 Ready 事件。store(true) 可以标记就绪，却不告诉你是不是第一次；先 load 再 store 又是两步。

swap 能够同时保存新值并返回旧值：

use std::sync::atomic::{AtomicBool, Ordering};

fn main() {
    let ready = AtomicBool::new(false);
    let first = ready.swap(true, Ordering::SeqCst);
    let second = ready.swap(true, Ordering::SeqCst);
    assert_eq!(first, false);
    assert_eq!(second, true);
}

首次旧值 false，表示此前没处理；第二次旧值 true，说明处理过了。项目把这个判断写成：

if launch_line_is_ready(&line) && !ready.swap(true, Ordering::SeqCst) {
    let _ = state_tx.send(LaunchEvent::Ready);
}

先检查日志是否匹配；只有匹配才执行右边。swap 把 ready 设为 true，! 将旧值取反。旧值 false 时，条件成立，发送 Ready，保证就绪分支至多进入一次。

2.6 compare_exchange：只有旧值匹配时才更新

swap 方法是无条件写入的。有时我们希望“只有值为 Idle 才改成 Running”，可以使用 compare_exchange：比较期望旧值，匹配才保存新值。

use std::sync::atomic::{AtomicUsize, Ordering};

fn main() {
    let state = AtomicUsize::new(0); // 约定：0 是 Idle，1 是 Running。
    assert_eq!(state.compare_exchange(0, 1, Ordering::SeqCst, Ordering::SeqCst), Ok(0));
    assert_eq!(state.compare_exchange(0, 1, Ordering::SeqCst, Ordering::SeqCst), Err(1));
}

第一次期望 0，实际也是 0，写入 1 并返回 Ok(旧值)。第二次仍期望 0，实际为 1，不修改，返回 Err(观察值)。

两个 Ordering 分别对应成功和失败时的顺序选择。我们将马上进入对它的介绍。

3 原子操作顺序 Ordering

3.1 Release / Acquire：先存、后取

现有一个任务：写线程修改一个原子值 value 后，读线程需要及时获取修改后的值。

一种想法是再引入一个原子 Bool 值 ready，初始值为 false。当写线程做出修改后，把 ready 设为 true。当读线程“看见” true 后，再执行对 value 的读取。

为了确保 ready 总是在 value 之后做出修改，value 的值总是在 ready 存储之后读取，我们对原子操作引入内存顺序的概念。

来认识 Release/Acquire 这一对 Ordering：

• Release 与 store 结合时，其之前的原子操作总是能够排在该 store 的顺序之前；
• Acquire 与 load 结合时，如果加载的值采用了 Release 顺序，则所有后续操作都将在该 load 操作之后。

use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
use std::thread;

fn main() {
    let value = AtomicUsize::new(0);
    let ready = AtomicBool::new(false);
    thread::scope(|scope| {
        scope.spawn(|| { // 写线程
            value.store(42, Ordering::Relaxed);
            ready.store(true, Ordering::Release);
        });
        scope.spawn(|| { // 读线程
            while !ready.load(Ordering::Acquire) {
                thread::yield_now();
            }
            assert_eq!(value.load(Ordering::Relaxed), 42);
        });
    });
}

thread::scope 允许线程借用外面的变量，并在返回前等它们结束；spawn 创建线程；yield_now 在示例循环里让出执行机会。需要注意，项目的复杂等待通常用锁或通道，不建议直接照搬这个循环。

3.2 AcqRel：一次操作既接收，也发布

上面的读线程只接收数据，写线程只发布数据。如果一个线程既要接收前一个线程的结果，又要把自己准备的结果交给后一个线程，该怎么表达？

先回忆 swap：它读取旧值，再写入新值，这两部分组成一次不可分割的原子操作。AcqRel 就用于这类同时包含读和写的操作：读的部分使用 Acquire，写的部分使用 Release。

下面把 AZULC 整合包安装功能中的“下载阶段→校验阶段”简化成一个教学例子。约定一个原子整数 stage：0 表示尚未下载完成，1 表示下载结果可读，2 表示下一阶段的数据可读。中间线程希望在确认 stage 仍为 1 时，把它改成 2，因此使用 compare_exchange：旧值符合预期才写入新值，成功返回 Ok，失败返回 Err。

// 教学节选：三个线程共用的原子变量。
// stage、downloaded、checked 初始均为 0；每个数据变量只有一个写者。

// 线程 A：保存下载结果，然后发布阶段 1。
downloaded.store(42, Ordering::Relaxed);
stage.store(1, Ordering::Release);

// 线程 B：独立修改 checked，不依赖 downloaded。
checked.store(7, Ordering::Relaxed);
let result = stage.compare_exchange(
    1, 2, Ordering::AcqRel, Ordering::Relaxed,
);
if result.is_ok() {
    assert_eq!(downloaded.load(Ordering::Relaxed), 42);
}

// 线程 C：观察到阶段 2，才使用交接的数据。
if stage.load(Ordering::Acquire) == 2 {
    assert_eq!(downloaded.load(Ordering::Relaxed), 42);
    assert_eq!(checked.load(Ordering::Relaxed), 7);
}

这里的三个线程可以交错执行，放在一起只是方便阅读。

先只看 compare_exchange 成功的情况，把这次操作的两部分分开理解：

这条同步关系还把 A 的结果传到了 C：A 先发布给 B，B 再发布给 C，所以 C 在上述条件下也能读到 42。

如果 B 执行时 stage 还是 0，比较就失败：不会写入 2，C 也不会进入读取分支。第四个参数 Relaxed 是失败时的读取顺序。

3.3 SeqCst：把两个线程的操作放在一起看

SeqCst 是 Sequentially Consistent 的缩写，通常译作“顺序一致性”。

假设启动器里有两个工作线程，各自记录“我已经开始”，随后检查对方有没有开始。两个标志初始都是 false，每个标志只会被写入一次 true：

线程 A	线程 B
A1：把 a_started 写成 true	B1：把 b_started 写成 true
A2：读取 b_started	B2：读取 a_started

如果这四次访问都使用 SeqCst，就可以把它们排成一份先后顺序来解释结果。排列必须保留各线程内部的顺序：A1 在 A2 前，B1 在 B2 前。但 A 和 B 谁先，并没有规定。

例如，以下两种顺序都允许：

现在问：两边能不能都读到 false？

在这个全用 SeqCst 的例子里，不能。可以一步步检查：

1. A2 如果读到 false，就必须排在 B1 写入 true 之前。
2. B1 在 B2 之前，这是 B 自己的代码顺序。
3. B2 如果也读到 false，就必须排在 A1 写入 true 之前。
4. 可 A1 又必须在 A2 之前，这是 A 自己的代码顺序。

这样就要求“A1 在 A2 前，A2 在 B1 前，B1 在 B2 前，B2 又在 A1 前”。这个顺序绕回了起点，排不出来，所以 SeqCst 排除了这个结果。

这就是“共同总顺序”想表达的意思：不同线程对这些操作的观察，必须能用同一份先后顺序解释，不能彼此矛盾。

如果四次访问都改用 Relaxed，就没有这份跨变量的共同顺序保证，两边都读到 false 是允许的。如果写入用 Release、读取用 Acquire，也仍允许这个结果，因为没有建立前面讲过的发布与接收关系。

3.4 总结一下

具体语义见 Rust Ordering 文档。