Rust 无锁编程实战：原子操作、内存排序与 Crossbeam 生产实践

1 阅读 0 评论 0 点赞

Rust 无锁编程实战：原子操作、内存排序与 Crossbeam 生产实践

现代多核系统对并发性能的要求越来越高，互斥锁在高竞争场景下会成为严重瓶颈。无锁（lock-free）数据结构在 Rust 中是一门需要谨慎使用的技术，但它也能带来数量级的性能提升。本文深入探讨 Rust 的原子类型、内存排序语义，以及如何构建稳健的无锁数据结构。

一、为什么需要无锁？互斥锁的真实代价

互斥锁（Mutex）的开发者友好性掩盖了其在高竞争下的性能问题。一个典型的互斥锁操作涉及以下开销：

系统调用开销：在 Linux 上，锁竞争会触发 futex() 系统调用，陷入内核态。一次 futex 系统调用耗时约 100-500 纳秒。
缓存一致性流量：当锁被释放并重新获取时，所有等待的 CPU 核心都会尝试读取锁变量，引发缓存行 bouncing。
调度器介入：竞争失败时线程可能被挂起，唤醒时需要重新调度。

对比之下，一个原子操作（如 fetch_add）在一条指令内完成，耗时约 10-20 纳秒，且不需要系统调用。

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

// 无锁计数器：8 线程并发递增 1000 万次各
fn lockfree_counter() -> u64 {
    let counter = Arc::new(AtomicU64::new(0));
    let mut handles = vec![];

    for _ in 0..8 {
        let c = Arc::clone(&counter);
        handles.push(thread::spawn(move || {
            for _ in 0..10_000_000 {
                c.fetch_add(1, Ordering::Relaxed);
            }
        }));
    }

    for h in handles {
        h.join().unwrap();
    }
    counter.load(Ordering::Relaxed)
}

这个例子看似简单，但 Ordering::Relaxed 的选择是有意为之——单纯的计数器确实只需要 relaxation，暂时不需要更强的排序保证。

二、Rust 的原子类型体系

Rust 在 std::sync::atomic 中提供了完整的原子类型，涵盖以下分类：

类型	说明
`AtomicBool`	原子布尔值
`AtomicIsize` / `AtomicUsize`	原子指针宽度整数（常用作指针）
`AtomicI8` / `AtomicU8` 到 `AtomicI64` / `AtomicU64`	固定宽度原子整数
`AtomicPtr<T>`	原子裸指针

这些类型无一例外实现了 Sync，可以安全地跨线程共享。

原子操作的核心 API

// 基本操作
let val = atomic.load(Ordering::SeqCst);       // 读取
atomic.store(val, Ordering::SeqCst);           // 写入
let prev = atomic.swap(new_val, Ordering::SeqCst); // 交换

// Compare-and-Swap：无锁算法的核心原语
let result = atomic.compare_exchange(
    current,    // 期望值
    new,        // 新值
    Ordering::AcqRel, // 成功时的排序
    Ordering::Acquire, // 失败时的排序
);

// 更强的 CAS（避免伪失败）
let result = atomic.compare_exchange_weak(
    current, new,
    Ordering::AcqRel,
    Ordering::Acquire,
);

// 算术操作
let prev = atomic.fetch_add(1, Ordering::Relaxed);
let prev = atomic.fetch_sub(1, Ordering::Relaxed);
let prev = atomic.fetch_and(mask, Ordering::Relaxed);
let prev = atomic.fetch_or(bits, Ordering::Relaxed);

compare_exchange_weak 在某些平台上（如 ARM LL/SC 架构）可能发生"伪失败"（spurious failure），即在值匹配时也返回错误。但这在循环中使用时反而更高效，因为避免了双重检查的开销。

三、内存排序：并发编程中最微妙的部分

内存排序（Memory Ordering）是无锁编程中最容易出错的部分。Rust 采用了和 C++20 一致的内存模型。

四种排序语义

1. Relaxed（松弛排序）

// 最简单的排序：只保证原子性，不保证顺序
counter.fetch_add(1, Ordering::Relaxed);

Relaxed 操作只保证：

该原子操作的原子性（不会被撕裂）
对同一原子变量的修改在所有线程间有全序关系

但不保证：

其他内存操作的顺序可见性
偏序关系（happens-before）

适用场景：纯计数器、统计指标、引用计数（当引用计数变化不影响数据可见性时）。

2. Acquire / Release（获取/释放）

// 线程 A：释放
data.store(42, Ordering::Relaxed);
ready.store(true, Ordering::Release);  // 发布标记

// 线程 B：获取
while !ready.load(Ordering::Acquire) {  // 等待发布
    std::hint::spin_loop();
}
assert_eq!(data.load(Ordering::Relaxed), 42); // 保证可见

Release 保证：在此操作之前的所有内存写入，对执行了对应 Acquire 操作的线程可见。这建立了跨线程的 happens-before 关系。

核心规则：如果线程 B 通过 Acquire 读取到线程 A 通过 Release 写入的值，则 A 中 Release 之前的所有写入对 B 可见。

3. AcqRel（获取-释放）

// Read-Modify-Write 操作同时需要 Acquire 和 Release
atomic.fetch_sub(1, Ordering::AcqRel);

用于同时需要读取和修改的操作（如 fetch_sub），确保：

读取到之前的写入（Acquire 语义）
写结果对后续 Acquire 者可见（Release 语义）

4. SeqCst（顺序一致性）

// 最强排序：所有线程看到相同的操作顺序
atomic.fetch_add(1, Ordering::SeqCst);

SeqCst 保证一个全局单一修改顺序（Single Total Modification Order）——所有线程看到所有 SeqCst 操作以相同顺序发生。

这是最强的保证，也是最慢的。在 x86 上，SeqCst 写入比 Release 写入多一个 mfence 或 lock 前缀指令。

生产建议

┌──────────────────────────────────────────────────────────┐
│                  内存排序选择指南                          │
├──────────────┬───────────────────────────────────────────┤
│ Relaxed      │ 纯计数器、统计，无数据依赖                 │
│ Acquire/Release │ 标志位同步、互斥锁实现、通道通信      │
│ AcqRel       │ 引用计数增减、同时读写的 RMW 操作          │
│ SeqCst       │ 多个原子变量之间的全局顺序要求（少用）     │
└──────────────┴───────────────────────────────────────────┘

四、实战：构建无锁 Treiber Stack

Treiber Stack 是最经典的无锁数据结构，它是理解 CAS 循环和 ABA 问题的绝佳起点。

use std::sync::atomic::{AtomicPtr, Ordering};
use std::ptr;

struct Node<T> {
    data: T,
    next: *mut Node<T>,
}

pub struct LockFreeStack<T> {
    head: AtomicPtr<Node<T>>,
}

impl<T> LockFreeStack<T> {
    pub fn new() -> Self {
        LockFreeStack {
            head: AtomicPtr::new(ptr::null_mut()),
        }
    }

    pub fn push(&self, data: T) {
        let new_node = Box::into_raw(Box::new(Node {
            data,
            next: ptr::null_mut(),
        }));

        loop {
            let head = self.head.load(Ordering::Relaxed);
            unsafe { (*new_node).next = head; }

            match self.head.compare_exchange_weak(
                head,
                new_node,
                Ordering::Release,
                Ordering::Relaxed,
            ) {
                Ok(_) => break,
                // CAS 失败：head 已被其他线程修改，重试
                // 注意：next 指针已在 compare_exchange 中自动更新（因为我们是基于 head 的值）
                Err(_) => continue,
            }
        }
    }

    pub fn pop(&self) -> Option<T> {
        loop {
            let head = self.head.load(Ordering::Acquire)?;
            let next = unsafe { (*head).next };

            match self.head.compare_exchange_weak(
                head,
                next,
                Ordering::Release,
                Ordering::Relaxed,
            ) {
                Ok(_) => unsafe {
                    let node = Box::from_raw(head);
                    return Some(node.data);
                }
                Err(_) => continue,
            }
        }
    }
}

impl<T> Drop for LockFreeStack<T> {
    fn drop(&mut self) {
        let mut curr = *self.head.get_mut();
        while !curr.is_null() {
            let node = unsafe { Box::from_raw(curr) };
            curr = node.next;
            // node 在作用域结束时自动释放其 data
        }
    }
}

正确性分析

push 使用 compare_exchange_weak：在循环中，weak 版本在 ARM 上能避免不必要的 LL/SC 重试。
排序选择：
push 的 CAS 成功时使用 Release，确保新节点的内容对其他线程可见。
pop 的 load 使用 Acquire，确保读取 head 指针时能看到节点的完整内容。
内存安全：pop 出来的节点通过 Box::from_raw 重新获得所有权。

ABA 问题

考虑以下场景：

时刻1：线程 T1 读取 head = A, next = B
时刻2：线程 T2 执行 pop() pop() push(A)，head 又变回 A
时刻3：T1 执行 CAS(A, B)，成功！但 B 已被释放，导致 use-after-free

ABA 问题在无锁编程中很常见。解决方案包括：

Tagged Pointer（标签指针）：利用指针的未使用位（如低 2 位总是 0）存储标记。
hazard_pointer ：安全延迟释放（稍后介绍）。
Crossbeam epoch-based reclamation：工业级解决方案。

五、轻松搞定无锁内存回收：Epoch-Based Reclamation

在无锁数据结构中，你不能在读取指针的同时释放它——因为另一个线程可能正在访问这块内存。Crossbeam 的 epoch-based reclamation 是 Rust 生态中最成熟的安全内存回收方案。

use crossbeam_epoch::{self as epoch, Atomic, Owned, Shared, Guard};
use std::sync::atomic::Ordering;

struct TreiberNode<T> {
    data: T,
    next: Atomic<TreiberNode<T>>,
}

pub struct EpochStack<T> {
    head: Atomic<TreiberNode<T>>,
}

impl<T> EpochStack<T> {
    pub fn push(&self, data: T) {
        let guard = epoch::pin();
        let new_node = Owned::new(TreiberNode {
            data,
            next: Atomic::null(),
        });

        loop {
            let head = self.head.load(Ordering::Relaxed, &guard);
            new_node.next.store(head, Ordering::Relaxed);

            match self.head.compare_exchange(
                head,
                new_node,
                Ordering::Release,
                Ordering::Relaxed,
                &guard,
            ) {
                Ok(_) => break,
                Err(_) => continue,
            }
        }
    }

    pub fn pop(&self) -> Option<T> {
        let guard = epoch::pin();

        loop {
            let head = self.head.load(Ordering::Acquire, &guard)?;
            let next = head.next.load(Ordering::Relaxed, &guard);

            if self.head.compare_exchange(
                head,
                next,
                Ordering::Release,
                Ordering::Relaxed,
                &guard,
            ).is_ok() {
                // 延迟释放：不是现在释放，而是在安全时机释放
                unsafe {
                    guard.defer_destroy(head);
                    return Some(head.as_ref().unwrap().data);
                }
            }
        }
    }
    }

}

关键要点：

epoch::pin() 获取当前线程的 epoch 标记。当进入一个 epoch 时，该线程正在访问的内存不会被回收。
延迟释放通过 guard.defer_destroy(head)。只有当所有线程都退出当前 epoch 后，内存才会真正释放。
这样我们可以完全避免 ABA 问题。

六、内存栅栏（Fence）：当原子操作不够时

有时你需要在非原子操作周围建立顺序保证，这时使用显式内存栅栏：

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

static DATA: AtomicBool = AtomicBool::new(false);

fn producer() {
    // 写入数据
    unsafe { DATA_PAYLOAD = 42u64; }

    // 释放栅栏：确保上面的写入在标记之前完成
    fence(Ordering::Release);

    DATA.store(true, Ordering::Relaxed);
}

fn consumer() {
    while !DATA.load(Ordering::Relaxed) {
        std::hint::spin_loop();
    }

    // 获取栅栏：确保标记读取后的操作能看到数据
    fence(Ordering::Acquire);

    assert_eq!(!(DATA_PAYLOAD), 42);
}

七、实战性能基准：Mutex vs Lock-Free

使用 Criterion.rs 对一个无锁队列做基准测试：

use criterion::{criterion_group, criterion_main, Criterion};
use crossbeam_seg_queue::SegQueue;
use std::sync::Mutex;

fn bench_queues(c: &mut Criterion) {
    let mut group = c.benchmark_group("queue_comparison");
    group.measurement_time(std::time::Duration::from_secs(5));

    group.bench_function("mutex_vecdeque", |b| {
        let q = std::sync::Arc::new(Mutex::new(std::collections::VecDeque::new()));
        b.iter(|| {
            let mut q = q.lock().unwrap();
            q.push_back(1u64);
            q.pop_front();
        });
    });

    group.bench_function("seg_queue", |b| {
        let q = std::sync::Arc::new(SegQueue::new());
        b.iter(|| {
            q.push(1u64);
            q.pop();
        });
    });
    group.finish();
}

criterion_group!(benches, bench_queues);
criterion_main!(benches);

八、生产建议

从标准库和成熟库开始：crossbeam 提供了无锁队列、SkipList、SegQueue 等优质实现，不要轻易自己实现无锁数据结构。

性能并非万能：无锁不一定更快。在低竞争或无竞争时，Mutex 通常更快。

测试要充分：使用 loom 进行并发模型验证，使用 MIRI 进行未定义行为检测。

# 在 CI 中加入 Loom 测试
MIRIFLAGS="-Zmiri-disable-isolation" cargo +nightly test --features loom

谨慎使用 Relaxed：确保你真的不需要跨线程同步。多审查、多测。

ABA 防护推荐使用 Hazard Pointers / Epochs：Crossbeam 的 epoch-based reclamation 是 Rust 生态中最成熟的解决方案。

总结

Rust 的无锁编程生态已经非常成熟。std::sync::atomic 提供基础设施，crossbeam 提供高级数据结构，loom 提供验证工具。无锁编程仍然是系统编程中最困难的部分之一，但在 Rust 的世界中，它从未如此安全。

点赞(0) 打赏

本文分类：AI应用开发
本文标签：Rust Lock-Free Concurrency Systems Programming Atomics Memory Ordering
浏览次数：1 次浏览
发布日期：2026-10-01 06:21:22
本文链接：https://www.ybb.press/ai-applications/10483.html

Rust 无锁编程实战：原子操作、内存排序与 Crossbeam 生产实践