Lock-Free 数据结构设计艺术

在高并发系统中，锁竞争是性能瓶颈的主要来源。Lock-Free 数据结构通过 CAS（Compare-And-Swap）原子操作和精心设计的内存序，实现无锁的线程安全访问。本文将深入实现无锁队列、无锁栈和无锁哈希表，并附正确性证明。

一、CAS 与 ABA 问题

CAS 是 Lock-Free 编程的基石。但 CAS 存在经典的 ABA 问题：线程 1 读取值 A，线程 2 将 A 改为 B 再改回 A，线程 1 的 CAS 仍然成功但语义已不正确。解决方案是使用带版本号的指针。

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

struct TaggedPtr {
    ptr: *mut T,
    tag: usize,
}

impl TaggedPtr {
    fn pack(ptr: *mut T, tag: usize) -> u128 {
        ((tag as u128) << 64) | (ptr as u128)
    }

    fn unpack(packed: u128) -> Self {
        TaggedPtr {
            ptr: (packed & 0xFFFFFFFFFFFFFFFF) as *mut T,
            tag: (packed >> 64) as usize,
        }
    }
}

二、无锁队列（Michael-Scott Queue）

经典的 Michael-Scott 无锁队列使用两个原子指针：Head 和 Tail。入队操作在 Tail 节点后追加新节点，出队操作从 Head 取出。关键技巧是两阶段 CAS：先链接新节点，再更新 Tail。

struct Node {
    value: Option,
    next: AtomicPtr>,
}

pub struct LockFreeQueue {
    head: AtomicPtr>,
    tail: AtomicPtr>,
}

impl LockFreeQueue {
    pub fn enqueue(&self, value: T) {
        let node = Box::into_raw(Box::new(Node {
            value: Some(value),
            next: AtomicPtr::new(ptr::null_mut()),
        }));

        loop {
            let tail = self.tail.load(Ordering::Acquire);
            let next = unsafe { (*tail).next.load(Ordering::Acquire) };

            if tail == self.tail.load(Ordering::Acquire) {
                if next.is_null() {
                    if unsafe { (*tail).next.compare_exchange_weak(
                        next, node, Ordering::Release, Ordering::Relaxed
                    ).is_ok() } {
                        self.tail.compare_exchange_weak(
                            tail, node, Ordering::Release, Ordering::Relaxed
                        ).ok();
                        return;
                    }
                } else {
                    self.tail.compare_exchange_weak(
                        tail, next, Ordering::Release, Ordering::Relaxed
                    ).ok();
                }
            }
        }
    }

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

            if head == self.head.load(Ordering::Acquire) {
                if head == tail {
                    if next.is_null() { return None; }
                    self.tail.compare_exchange_weak(
                        tail, next, Ordering::Release, Ordering::Relaxed
                    ).ok();
                } else {
                    let value = unsafe { (*next).value.take() };
                    if self.head.compare_exchange_weak(
                        head, next, Ordering::Release, Ordering::Relaxed
                    ).is_ok() {
                        unsafe { drop(Box::from_raw(head)); }
                        return value;
                    }
                }
            }
        }
    }
}

三、无锁栈（Treiber Stack）

Treiber Stack 是最简单的无锁数据结构，只需一个原子指针指向栈顶。Push 和 Pop 都使用单次 CAS 操作，实现极其简洁。

pub struct LockFreeStack {
    head: AtomicPtr>,
}

impl LockFreeStack {
    pub fn push(&self, value: T) {
        let node = Box::into_raw(Box::new(Node { value, next: AtomicPtr::new(ptr::null_mut()) }));
        loop {
            let head = self.head.load(Ordering::Relaxed);
            unsafe { (*node).next.store(head, Ordering::Relaxed); }
            if self.head.compare_exchange_weak(
                head, node, Ordering::Release, Ordering::Relaxed
            ).is_ok() { break; }
        }
    }

    pub fn pop(&self) -> Option {
        loop {
            let head = self.head.load(Ordering::Acquire);
            if head.is_null() { return None; }
            let next = unsafe { (*head).next.load(Ordering::Relaxed) };
            if self.head.compare_exchange_weak(
                head, next, Ordering::Release, Ordering::Relaxed
            ).is_ok() {
                let value = unsafe { ptr::read(&(*head).value) };
                unsafe { drop(Box::from_raw(head)); }
                return Some(value);
            }
        }
    }
}

四、内存序模型精要

C++11/Rust 内存序模型定义了 6 种顺序：Relaxed、Acquire、Release、AcqRel、SeqCst 和 Consume。正确选择内存序是 Lock-Free 编程中最容易出错的部分。核心原则：写操作使用 Release，读操作使用 Acquire，形成 happens-before 关系。

五、正确性证明方法

Lock-Free 数据结构的正确性可以通过线性一致性（Linearizability）来证明：每个操作在其调用和返回之间的某个点瞬间生效。使用 LP（Linearization Point）方法，为每个操作确定一个原子生效的时间点。

掌握 Lock-Free 编程是成为系统级工程师的关键技能，在高频交易、数据库内核、游戏引擎等场景中不可或缺。