rust原子操作与无锁编程

1. 原子操作概念

原子操作是指对某一个原子变量执行的不可分割读/写/读改写操作。在多线程环境下，其他线程不会看到这个操作的“中间状态”。注意：多个原子操作连在一起并不会自动变成一个整体原子事务；如果需要维护多个变量之间的不变量，通常仍然需要锁或精心设计的无锁算法。CPU 层面会通过特殊指令（如 CAS、XADD）或等价机制来保证单次原子操作的原子性。

Rust标准库在 std::sync::atomic模块中提供了与平台相关的原子类型，例如：

• AtomicBool
• AtomicI8 / AtomicI16 / AtomicI32 / AtomicI64 / AtomicIsize
• AtomicU8 / AtomicU16 / AtomicU32 / AtomicU64 / AtomicUsize
• AtomicPtr<T>

这些类型的方法通常接收一个内存序（Ordering）参数，控制操作的可见性和重排序行为。

2. 原子操作与锁的区别

简单记忆：锁适合大粒度、复杂操作；原子操作适合小粒度、高性能计数/标志。

3. 原子操作基本使用

3.1 创建原子变量 – AtomicXXX::new

功能：创建一个新的原子变量，初始化为给定值。

接口签名（以 AtomicUsize为例）：

#![allow(unused)]
fn main() {
pub const fn new(v: usize) -> AtomicUsize
}

• 参数：v – 初始值。
• 返回值：AtomicUsize实例。

简单调用示例：

#![allow(unused)]
fn main() {
use std::sync::atomic::AtomicUsize;
let counter = AtomicUsize::new(0);
}

3.2 加载值 – load

功能：原子地读取当前值，不可与写操作交错。

接口签名：

#![allow(unused)]
fn main() {
pub fn load(&self, order: Ordering) -> usize
}

• 参数：order – 内存序（见第四节）。
• 返回值：当前值。

简单调用示例：

#![allow(unused)]
fn main() {
let value = counter.load(std::sync::atomic::Ordering::SeqCst);
}

3.3 存储值 – store

功能：原子地写入新值，覆盖旧值。

接口签名：

#![allow(unused)]
fn main() {
pub fn store(&self, val: usize, order: Ordering)
}

• 参数：
  • val – 要写入的值。
  • order – 内存序。
• 返回值：无。

简单调用示例：

#![allow(unused)]
fn main() {
counter.store(42, std::sync::atomic::Ordering::SeqCst);
}

3.4 交换值 – swap

功能：原子地将新值写入，并返回原来的旧值（读-改-写原子操作）。

接口签名：

#![allow(unused)]
fn main() {
pub fn swap(&self, val: usize, order: Ordering) -> usize
}

• 参数：
  • val – 新值。
  • order – 内存序。
• 返回值：旧值。

简单调用示例：

#![allow(unused)]
fn main() {
let old = counter.swap(100, std::sync::atomic::Ordering::SeqCst);
}

3.5 获取并修改（fetch操作族）

3.5.1 fetch_add / fetch_sub

功能：原子地将当前值加上（或减去）一个值，返回修改前的值。

接口签名：

#![allow(unused)]
fn main() {
pub fn fetch_add(&self, val: usize, order: Ordering) -> usize
}

• 参数：
  • val – 要加的值。
  • order – 内存序。
• 返回值：修改前的值。

简单调用示例：

#![allow(unused)]
fn main() {
let previous = counter.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
}

类似的还有 fetch_and, fetch_or, fetch_xor, fetch_max, fetch_min等。

4. 内存序（Memory Ordering）

内存序定义了原子操作的可见性和重排序约束，用于平衡性能和正确性。Rust使用 std::sync::atomic::Ordering枚举，包含以下变体：

4.1 Relaxed（松散序）

• 保证：单个原子操作是原子的，但不对其他操作的顺序做任何保证。
• 用途：仅需要原子性，不关心同步（例如仅用作计数器，不加锁统计次数）。

4.2 Release（释放） & Acquire（获取）

• Release：禁止当前线程中它之前的读写被重排到该操作之后，并把之前的写入“发布”出去。
• Acquire：禁止当前线程中它之后的读写被重排到该操作之前；如果它读取到了某个 Release 操作写入的值，就与那个 Release 建立同步关系，从而能看到对方在 Release 之前的写入。
• 用途：实现锁、传递数据的所有权（例如 Mutex的内部实现）。

4.3 AcqRel（获取-释放）

• 保证：同时具有 Acquire和 Release的语义，通常用于读-改-写操作（如 swap, fetch_add, CAS）。对读改写操作来说，成功路径才同时具备读和写的同步语义；失败路径通常只按失败内存序处理。
• 用途：同时需要读和写的操作。

4.4 SeqCst（顺序一致）

• 保证：最强的常用约束，所有 SeqCst 原子操作参与同一个全局顺序，便于推理。
• 用途：简单推理，性能有开销，但当不确定时首选。

4.5 内存序对比示例

#![allow(unused)]
fn main() {
use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
use std::thread;

static FLAG: AtomicBool = AtomicBool::new(false);
static DATA: AtomicUsize = AtomicUsize::new(0);

// 线程1：写数据，然后设置标志
thread::spawn(move || {
    DATA.store(42, Ordering::Relaxed);     // 原子写入数据本身，用 Relaxed 即可
    FLAG.store(true, Ordering::Release);   // 释放屏障
});

// 线程2：等待标志，然后读数据
thread::spawn(move || {
    while !FLAG.load(Ordering::Acquire) {} // 获取屏障
    let data = DATA.load(Ordering::Relaxed);
    assert_eq!(data, 42); // 因为Release-Acquire同步，保证看到42
});
}

5. CAS操作（Compare And Swap）

CAS是“比较并交换”的缩写，是大多数无锁数据结构的基石。

5.1 compare_exchange

功能：原子地比较当前值与 current是否相等；若相等则设置为 new，否则不做修改。返回结果并告知是否成功。

接口签名：

#![allow(unused)]
fn main() {
pub fn compare_exchange(&self, current: usize, new: usize, success: Ordering, failure: Ordering) -> Result<usize, usize>
}

• 参数：
  • current – 期望的当前值。
  • new – 要设置的新值。
  • success – 当CAS成功时的内存序。
  • failure – 当CAS失败时的内存序（不能强于 success，且不能是 Release或 AcqRel）。
• 返回值：
  • Ok(old) – 成功，old等于 current，值已被更新。
  • Err(old) – 失败，old是实际的当前值。

简单调用示例：

#![allow(unused)]
fn main() {
let mut current = counter.load(Ordering::SeqCst);
loop {
    match counter.compare_exchange(current, current + 1, Ordering::SeqCst, Ordering::SeqCst) {
        Ok(_) => break,
        Err(actual) => current = actual, // 更新期望值，重试
    }
}
}

5.2 compare_exchange_weak

功能：与 compare_exchange类似，但允许在某些平台上虚假失败（即使值相等也返回 Err）。通常用于循环重试中，性能更高。

接口签名：

#![allow(unused)]
fn main() {
pub fn compare_exchange_weak(&self, current: usize, new: usize, success: Ordering, failure: Ordering) -> Result<usize, usize>
}

• 参数、返回值：同 compare_exchange。
• 注意：必须放在循环中，因为虚假失败时需重试。

简单调用示例：

#![allow(unused)]
fn main() {
let mut current = counter.load(Ordering::SeqCst);
loop {
    match counter.compare_exchange_weak(current, current + 1, Ordering::SeqCst, Ordering::SeqCst) {
        Ok(_) => break,
        Err(actual) => current = actual,
    }
}
}

6. ABA问题及其解决方法

6.1 什么是ABA问题

假设有一个共享变量 V，值为 A。线程T1读取 V得到 A，然后被操作系统挂起。线程T2将 V改为 B，再改回 A。当T1恢复执行并执行CAS(A→新值)时，会成功，因为当前值仍然是 A。但T1不知道 V已经经历过 A→B→A的变化，这可能导致数据不一致（例如指针指向的内存已被释放又重新分配）。

6.2 解决ABA问题的方法

常见的解决方案是使用带有版本号的指针或双字CAS。思路是：CAS 时不只比较指针本身，还比较一个随修改递增的版本号。

由于 Rust 标准库没有提供通用的双字 CAS（需要平台支持），下面用 AtomicUsize 演示“指针 + 版本号”打包思路：假设平台指针只使用低 48 位，高位用于保存版本号。这个假设不是可移植的生产方案，只适合帮助理解 ABA 问题。

6.3 模拟带版本号的CAS示例

下面不用裸指针，而是用“槽位索引 + 版本号”演示 ABA 的解决思路。

假设低 16 位保存槽位索引，高位保存版本号。每次更新索引时，版本号都加 1。这样即使索引从 A -> B -> A 回到原值，版本号也已经变化，旧的 CAS 就不会误判成功。

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

const INDEX_BITS: usize = 16;
const INDEX_MASK: usize = (1 << INDEX_BITS) - 1;

fn pack(version: usize, index: usize) -> usize {
    (version << INDEX_BITS) | (index & INDEX_MASK)
}

fn unpack(value: usize) -> (usize, usize) {
    let version = value >> INDEX_BITS;
    let index = value & INDEX_MASK;
    (version, index)
}

struct VersionedIndex {
    state: AtomicUsize,
}

impl VersionedIndex {
    fn new(index: usize) -> Self {
        Self {
            state: AtomicUsize::new(pack(0, index)),
        }
    }

    fn load(&self) -> (usize, usize) {
        unpack(self.state.load(Ordering::SeqCst))
    }

    fn update_index(&self, expected: (usize, usize), new_index: usize) -> bool {
        let expected_raw = pack(expected.0, expected.1);
        let new_raw = pack(expected.0 + 1, new_index);

        self.state
            .compare_exchange(expected_raw, new_raw, Ordering::SeqCst, Ordering::SeqCst)
            .is_ok()
    }
}

fn main() {
    let index = VersionedIndex::new(7);

    let old = index.load();
    assert_eq!(old, (0, 7));

    assert!(index.update_index(old, 9));
    assert_eq!(index.load(), (1, 9));

    // 即使试图把索引改回 7，版本号也会继续增加。
    assert!(index.update_index((1, 9), 7));
    assert_eq!(index.load(), (2, 7));

    // 旧观察值 (0, 7) 已经过期，CAS 不会成功。
    assert!(!index.update_index(old, 10));
}

注：这个例子只演示版本号如何避免 ABA 误判。真实无锁数据结构还要处理内存回收、缓存行竞争、平台位宽等问题，生产环境建议使用成熟库。

7. 完整代码示例：SPSC 有界无锁队列

SPSC 是 Single Producer Single Consumer 的缩写，表示“单生产者、单消费者”。

这个约束非常关键：

• 只有一个生产者线程调用 push。
• 只有一个消费者线程调用 pop。
• 生产者只负责写入 tail 指向的槽位。
• 消费者只负责读取 head 指向的槽位。

如果想做一个通用的 SpscQueue<T>，通常需要处理“槽位尚未初始化”和“从槽位中移动出 T”的问题，标准库教学代码会变得明显更复杂。为了让本章示例保持完全安全，这里实现一个受限但安全的版本：

• 队列元素固定为 i32。
• 每个槽位用 AtomicI32 存储。
• 保留 i32::MIN 作为空槽位哨兵值，因此不能发送这个值。

这个版本不适合当作通用生产级队列，但很适合学习原子变量、环形队列和 Acquire/Release 同步。

7.1 设计思路

这个队列使用固定大小的环形缓冲区：

buffer: [ AtomicI32, AtomicI32, AtomicI32, ... ]
head: 下一个要读的位置
tail: 下一个要写的位置

判断规则：

• head == tail：队列为空。
• next_tail == head：队列已满。

为了区分“空”和“满”，环形队列通常会空出一个槽位，所以如果内部数组长度是 N，实际可用容量是 N - 1。

7.2 内存序选择

这个例子主要使用 Acquire / Release：

• 生产者写入槽位后，用 tail.store(..., Ordering::Release) 发布新元素。
• 消费者用 tail.load(Ordering::Acquire) 观察生产者发布的新元素。
• 消费者取走数据并清空槽位后，用 head.store(..., Ordering::Release) 发布空槽位。
• 生产者用 head.load(Ordering::Acquire) 观察消费者释放出来的空槽位。

Relaxed 用在“只由当前线程修改”的索引读取上：

• tail 只由生产者修改，所以生产者读取自己的 tail 可以用 Relaxed。
• head 只由消费者修改，所以消费者读取自己的 head 可以用 Relaxed。

槽位本身也是原子变量。槽位的 Relaxed 写入和读取，依靠 tail/head 上的 Release/Acquire 建立跨线程可见性。

7.3 代码示例

#![allow(unused)]
fn main() {
use std::array;
use std::sync::Arc;
use std::sync::atomic::{AtomicI32, AtomicUsize, Ordering};
use std::thread;

const EMPTY: i32 = i32::MIN;

// 一个完全使用安全 Rust 写成的教学版 SPSC 无锁队列。
//
// 为了做到完全安全，这里不存放泛型 T，也不使用未初始化内存，
// 而是把每个槽位设计成 AtomicI32，并保留 i32::MIN 作为“空槽位”哨兵值。
// 这让示例更适合学习原子操作，但也意味着它不是通用队列。
pub struct SpscQueue<const N: usize> {
    // 每个槽位都是一个原子 i32。EMPTY 表示该槽位当前没有有效数据。
    buffer: [AtomicI32; N],
    // head 指向下一个要读取的位置，只由消费者线程更新。
    head: AtomicUsize,
    // tail 指向下一个要写入的位置，只由生产者线程更新。
    tail: AtomicUsize,
}

impl<const N: usize> SpscQueue<N> {
    pub fn new() -> Self {
        assert!(N > 1, "SPSC 队列容量必须大于 1");

        Self {
            buffer: array::from_fn(|_| AtomicI32::new(EMPTY)),
            head: AtomicUsize::new(0),
            tail: AtomicUsize::new(0),
        }
    }

    fn next(index: usize) -> usize {
        (index + 1) % N
    }

    // 实际可用容量是 N - 1。
    // 空出一个槽位后，head == tail 表示空，next_tail == head 表示满。
    pub fn capacity(&self) -> usize {
        N - 1
    }

    pub fn push(&self, value: i32) -> Result<(), i32> {
        assert_ne!(value, EMPTY, "i32::MIN 被保留为空槽位哨兵值");

        // tail 只由生产者修改，生产者读自己的 tail 用 Relaxed 即可。
        let tail = self.tail.load(Ordering::Relaxed);
        let next_tail = Self::next(tail);

        // head 由消费者发布。Acquire 保证生产者能看到消费者释放的槽位。
        let head = self.head.load(Ordering::Acquire);
        if next_tail == head {
            return Err(value);
        }

        // 写入数据本身。这个写入会被下面的 tail.store(Release) 发布出去。
        self.buffer[tail].store(value, Ordering::Relaxed);

        // Release 发布新的 tail。
        // 消费者用 Acquire 读到这个 tail 值后，就能看到上面对槽位的写入。
        self.tail.store(next_tail, Ordering::Release);
        Ok(())
    }

    pub fn pop(&self) -> Option<i32> {
        // head 只由消费者修改，消费者读自己的 head 用 Relaxed 即可。
        let head = self.head.load(Ordering::Relaxed);

        // tail 由生产者发布。Acquire 保证消费者能看到生产者写入的数据。
        let tail = self.tail.load(Ordering::Acquire);
        if head == tail {
            return None;
        }

        let value = self.buffer[head].load(Ordering::Relaxed);
        debug_assert_ne!(value, EMPTY);

        // 把槽位重新标记为空。这个写入会被下面的 head.store(Release) 发布出去。
        self.buffer[head].store(EMPTY, Ordering::Relaxed);

        // Release 发布新的 head，让生产者知道这个槽位可以复用。
        self.head.store(Self::next(head), Ordering::Release);
        Some(value)
    }

    pub fn is_empty(&self) -> bool {
        self.head.load(Ordering::Acquire) == self.tail.load(Ordering::Acquire)
    }
}

pub fn demo() {
    println!("...............原子操作 SPSC 无锁队列示例开始.................");

    let queue = Arc::new(SpscQueue::<8>::new());
    println!("队列可用容量: {}", queue.capacity());

    let producer_queue = Arc::clone(&queue);
    let producer = thread::spawn(move || {
        for value in 0..20 {
            let mut value = value;
            loop {
                match producer_queue.push(value) {
                    Ok(()) => break,
                    Err(returned) => {
                        // 队列满了，拿回发送失败的值，稍后重试。
                        value = returned;
                        thread::yield_now();
                    }
                }
            }
        }

        // 用 -1 作为结束标记，告诉消费者可以退出。
        let mut end = -1;
        while let Err(returned) = producer_queue.push(end) {
            end = returned;
            thread::yield_now();
        }
    });

    let consumer_queue = Arc::clone(&queue);
    let consumer = thread::spawn(move || {
        let mut count = 0;
        let mut sum = 0;

        loop {
            match consumer_queue.pop() {
                Some(-1) => break,
                Some(value) => {
                    count += 1;
                    sum += value;
                    println!("consumer pop: {}", value);
                }
                None => thread::yield_now(),
            }
        }

        (count, sum)
    });

    producer.join().unwrap();
    let (count, sum) = consumer.join().unwrap();
    println!("消费者共收到 {} 个数据，求和结果 {}", count, sum);
    println!("队列是否为空: {}", queue.is_empty());

    println!("...............原子操作 SPSC 无锁队列示例结束.................");
}
}

7.4 代码讲解

生产者写入元素的顺序是：

1. 读取 tail，找到要写入的槽位。
2. 读取 head，确认队列没有满。
3. 把数据写入 buffer[tail]。
4. 用 Release 更新 tail，把新元素发布给消费者。

消费者读取元素的顺序是：

1. 读取 head，找到要读取的槽位。
2. 用 Acquire 读取 tail，确认生产者已经发布了元素。
3. 从 buffer[head] 读取数据。
4. 把 buffer[head] 重新写回 EMPTY。
5. 用 Release 更新 head，把空槽位发布给生产者复用。

这个示例能保持完全安全，原因是：

• 队列不存放泛型 T，只存放 AtomicI32。
• 所有槽位从一开始就初始化为 EMPTY。
• 读取和写入槽位都通过安全的原子 API 完成。

7.5 使用限制

这个示例是教学版 SPSC 队列，不是通用队列：

• 不能有多个生产者同时调用 push。
• 不能有多个消费者同时调用 pop。
• 只能存放 i32。
• 不能存放 i32::MIN，因为它被保留为空槽位哨兵值。
• 队列是固定容量，满了以后 push 会返回 Err(value)。
• 如果要实现通用的 SpscQueue<T> 或 MPMC 无锁队列，建议使用成熟库，例如 crossbeam。