无锁栈：Treiber 栈与指数退避

1986 年，R. Kent Treiber 在 IBM 的一份技术报告中描述了一个极其简洁的无锁栈实现。整个算法只有两个操作——push 和 pop——每个操作的核心都是一次 CAS（Compare-And-Swap）。

三十多年后的今天，Treiber 栈仍然是并发数据结构的”Hello World”——每一本并发编程教材都会从它讲起。不是因为它最实用（它有严重的可扩展性问题），而是因为它浓缩了无锁编程的所有核心概念：CAS 循环、ABA 问题、内存回收、伪共享。

理解 Treiber 栈，就理解了无锁编程的基本范式。

下图展示了 Treiber 栈的 CAS 操作和 Elimination Backoff 的配对机制：

一、为什么需要无锁栈

锁的问题

一个简单的加锁栈：

typedef struct {
    Node *top;
    pthread_mutex_t lock;
} LockedStack;

void push(LockedStack *s, int val) {
    Node *n = malloc(sizeof(Node));
    n->val = val;
    pthread_mutex_lock(&s->lock);
    n->next = s->top;
    s->top = n;
    pthread_mutex_unlock(&s->lock);
}

问题在哪？

1. 优先级反转：低优先级线程持锁时，高优先级线程被阻塞
2. convoying：一个线程在临界区内被调度出去，所有其他线程排队等待
3. 不可组合：两个加锁操作不能安全组合（可能死锁）
4. cache line bouncing：锁变量本身的争抢导致 cache line 频繁失效

CAS：无锁的基石

CAS（Compare-And-Swap）是一条硬件原子指令：

// 原子操作：如果 *ptr == expected，则 *ptr = desired，返回 true
//           否则 expected = *ptr，返回 false
bool CAS(T *ptr, T *expected, T desired);

x86 上对应 CMPXCHG 指令，ARM 上对应 LDXR/STXR（LL/SC）对。

CAS 的核心优势：乐观并发。先假设没有竞争，直接尝试修改；如果失败（有人先修改了），重试。

二、Treiber 栈：最简单的无锁数据结构

算法描述

栈只有一个共享指针 top，指向栈顶节点。每个节点有 val 和 next。

Push： 1. 创建新节点 n 2. 读取当前 top 到 old_top 3. 设置 n->next = old_top 4. CAS(&top, &old_top, n) - 成功：done - 失败：回到第 2 步

Pop： 1. 读取当前 top 到 old_top 2. 如果 old_top == NULL，栈空 3. 读取 old_top->next 到 new_top 4. CAS(&top, &old_top, new_top) - 成功：返回 old_top->val - 失败：回到第 1 步

C 实现

#include <stdint.h>
#include <stdatomic.h>
#include <stdlib.h>
#include <stdio.h>

typedef struct Node {
    int val;
    struct Node *next;
} Node;

typedef struct {
    _Atomic(Node *) top;
} TreiberStack;

void treiber_init(TreiberStack *s) {
    atomic_store(&s->top, NULL);
}

void treiber_push(TreiberStack *s, int val) {
    Node *n = malloc(sizeof(Node));
    n->val = val;

    Node *old_top = atomic_load_explicit(&s->top, memory_order_relaxed);
    do {
        n->next = old_top;
    } while (!atomic_compare_exchange_weak_explicit(
        &s->top, &old_top, n,
        memory_order_release,   // 成功：release（让其他线程看到 n 的内容）
        memory_order_relaxed)); // 失败：relaxed（重试即可）
}

int treiber_pop(TreiberStack *s, int *val) {
    Node *old_top = atomic_load_explicit(&s->top, memory_order_acquire);
    Node *new_top;

    do {
        if (old_top == NULL) return 0;  // 栈空
        new_top = old_top->next;
    } while (!atomic_compare_exchange_weak_explicit(
        &s->top, &old_top, new_top,
        memory_order_acquire,
        memory_order_relaxed));

    *val = old_top->val;
    // 注意：这里不能直接 free(old_top)！
    // 其他线程可能还在读 old_top->next
    // 需要安全内存回收（Hazard Pointers 或 Epoch-Based）
    return 1;
}

内存序的选择

为什么 push 用 memory_order_release，pop 用 memory_order_acquire？

• Push 的 release：确保新节点 n 的 val 和 next 字段在 CAS 成功之前对其他线程可见。Release 保证之前的所有写操作不会被重排到 CAS 之后。

• Pop 的 acquire：确保读到 top 的值后，能看到该节点的 val 和 next。Acquire 保证之后的所有读操作不会被重排到加载 top 之前。

Release-acquire 对形成了 happens-before 关系：push 的 release 和 pop 的 acquire 同步，保证 pop 线程能看到 push 线程写入的数据。

在 x86 上，release 和 acquire 都是免费的（x86 的 TSO 内存模型天然提供）。但在 ARM 和 RISC-V 上，它们会生成额外的屏障指令。

三、ABA 问题：无锁编程的经典陷阱

问题描述

考虑以下场景，栈内容为 A → B → C：

线程 1                          线程 2
────────                        ────────
old_top = A
new_top = B
(被调度出去)
                                pop A → 成功
                                pop B → 成功
                                push A → A 回到栈顶
                                （此时栈：A → C）
(恢复执行)
CAS(&top, &A, B) → 成功！
(因为 top 确实是 A)

问题：线程 1 的 CAS 成功了，但 B 已经被释放了！top 现在指向已释放的 B。

这就是 ABA 问题：top 的值从 A 变成其他值，又变回 A。CAS 无法区分”没变”和”变了又变回来”。

Tagged Pointer 解法

最经典的解法：在指针旁边附带一个版本号（tag）。每次修改指针时，版本号加一。

// Tagged pointer: 高位存版本号，低位存指针
// 利用 x86 的 128-bit CAS (CMPXCHG16B)
typedef struct {
    Node *ptr;
    uintptr_t tag;
} TaggedPtr;

typedef struct {
    _Atomic(TaggedPtr) top;  // 需要 128-bit atomic
} TaggedStack;

void tagged_push(TaggedStack *s, int val) {
    Node *n = malloc(sizeof(Node));
    n->val = val;

    TaggedPtr old_top = atomic_load(&s->top);
    TaggedPtr new_top;
    do {
        n->next = old_top.ptr;
        new_top.ptr = n;
        new_top.tag = old_top.tag + 1;
    } while (!atomic_compare_exchange_weak(&s->top, &old_top, new_top));
}

int tagged_pop(TaggedStack *s, int *val) {
    TaggedPtr old_top = atomic_load(&s->top);
    TaggedPtr new_top;

    do {
        if (old_top.ptr == NULL) return 0;
        new_top.ptr = old_top.ptr->next;
        new_top.tag = old_top.tag + 1;
    } while (!atomic_compare_exchange_weak(&s->top, &old_top, new_top));

    *val = old_top.ptr->val;
    // 仍然需要安全回收 old_top.ptr
    return 1;
}

128-bit CAS 在 x86-64 上通过 CMPXCHG16B 支持（需要 16 字节对齐）。在 32 位系统上，64-bit CAS 就够用（32-bit 指针 + 32-bit tag）。

更好的方案：安全内存回收

Tagged pointer 解决了 ABA 问题，但没有解决内存回收问题——被 pop 出的节点什么时候能安全 free？

三种主要方案： - Hazard Pointers：每个线程发布正在访问的指针，回收时检查 - Epoch-Based Reclamation：全局 epoch 计数器，延迟到安全 epoch 后回收 - RCU：类似 epoch，但专门为读多写少优化

这三种方案在后续文章中详细讨论。

四、可扩展性问题：CAS 的瓶颈

为什么 Treiber 栈不可扩展

所有线程都在同一个 top 指针上做 CAS。当线程数增加时：

• CAS 失败率上升（更多竞争）
• Cache line 在核心间频繁弹跳（MESI 协议的 invalidation）
• 有效吞吐量反而下降

实测数据（Intel Xeon，push/pop 各 50%）：

注意：Treiber 栈在低竞争时（1-2 线程）明显优于 mutex，但在高竞争时（32 线程）差距几乎消失。

根本原因

LIFO 栈的语义决定了：所有操作必须线性化到同一个点（栈顶）。这是一个串行瓶颈（sequential bottleneck）。无论用什么并发技术，吞吐量都受限于 cache line 的传输延迟。

这不是实现的问题，而是数据结构定义的问题。

五、Elimination Backoff Stack：用并行化解决串行瓶颈

核心思想

2004 年，Hendler、Shavit 和 Yerushalmi 提出了 Elimination Backoff Stack。思路极其巧妙：

如果一个 push 和一个 pop 几乎同时发生，它们可以直接交换数据，不需要触碰栈本身。push 把元素给 pop，两者都完成——就好像 push 后立即 pop 了一样。

这相当于在栈旁边开了一个”消除数组”（elimination array）：

                ┌─────────────────┐
                │   Treiber 栈    │
                │   (串行瓶颈)    │
                └────────┬────────┘
                         │ 竞争激烈时
                ┌────────▼────────┐
                │  Elimination    │
                │  Array          │
                │ [slot0][slot1]..│
                │  push 和 pop    │
                │  在这里配对     │
                └─────────────────┘

算法流程

push(x):
  1. 尝试 CAS push 到栈
  2. 如果 CAS 失败（竞争激烈）:
     a. 随机选一个 elimination slot
     b. 把 x 写入 slot，等待 pop 来取
     c. 如果等到了 pop（slot 被清空）→ 成功
     d. 如果超时 → 回到第 1 步

pop():
  1. 尝试 CAS pop 从栈
  2. 如果 CAS 失败:
     a. 随机选一个 elimination slot
     b. 检查是否有 push 在等待
     c. 如果有 → 取走数据，清空 slot → 成功
     d. 如果没有 → 回到第 1 步

C 实现

#define ELIM_SIZE 8
#define ELIM_TIMEOUT_NS 1000  // 1 微秒超时

typedef enum { EMPTY, WAITING, BUSY } SlotState;

typedef struct {
    _Atomic(int) val;
    _Atomic(SlotState) state;
} EliminationSlot;

typedef struct {
    TreiberStack stack;
    EliminationSlot elim[ELIM_SIZE];
} EliminationStack;

int elim_push(EliminationStack *es, int val) {
    // 先尝试直接 push
    Node *n = malloc(sizeof(Node));
    n->val = val;
    Node *old_top = atomic_load(&es->stack.top);
    n->next = old_top;
    if (atomic_compare_exchange_strong(&es->stack.top, &old_top, n))
        return 1;

    // CAS 失败，尝试 elimination
    int slot = rand() % ELIM_SIZE;
    SlotState expected = EMPTY;
    atomic_store(&es->elim[slot].val, val);
    if (atomic_compare_exchange_strong(&es->elim[slot].state,
                                       &expected, WAITING)) {
        // 等待 pop 来配对
        struct timespec start;
        clock_gettime(CLOCK_MONOTONIC, &start);
        while (atomic_load(&es->elim[slot].state) == WAITING) {
            struct timespec now;
            clock_gettime(CLOCK_MONOTONIC, &now);
            if (elapsed_ns(start, now) > ELIM_TIMEOUT_NS) {
                // 超时，收回
                expected = WAITING;
                if (atomic_compare_exchange_strong(
                        &es->elim[slot].state, &expected, EMPTY)) {
                    free(n);
                    // 回到主栈重试
                    return elim_push(es, val);
                }
                // 如果 CAS 失败，说明 pop 已经来了
                break;
            }
        }
        free(n);  // 数据已通过 elimination 传递
        return 1;
    }

    free(n);
    return elim_push(es, val);  // 重试
}

性能特征

Elimination Stack 的神奇之处在于：竞争越激烈，elimination 越有效。push 和 pop 配对的概率随线程数增加而增加。

在 32 线程时，Elimination Stack 比 Treiber 快 22 倍——因为大部分操作通过 elimination 完成，不需要触碰共享的 top 指针。

六、指数退避：减少无效竞争

问题

当 CAS 失败时，最简单的策略是立即重试。但这会加剧竞争——所有失败的线程同时重试，再次碰撞。

指数退避算法

void push_with_backoff(TreiberStack *s, int val) {
    Node *n = malloc(sizeof(Node));
    n->val = val;

    int backoff = 1;  // 初始退避（微秒）
    int max_backoff = 1024;

    Node *old_top = atomic_load(&s->top);
    while (1) {
        n->next = old_top;
        if (atomic_compare_exchange_weak(&s->top, &old_top, n))
            return;

        // CAS 失败，退避
        int delay = backoff + (rand() % backoff);
        usleep(delay);
        backoff = (backoff * 2 < max_backoff) ? backoff * 2 : max_backoff;
    }
}

指数退避将 CAS 重试的间隔按 1, 2, 4, 8, … 倍增长，减少同时重试的线程数。以太网的 CSMA/CD 就用了类似的策略。

退避 vs Elimination

七、Flat Combining：另一种思路

从竞争到协作

2010 年，Hendler 等人提出了 Flat Combining。思路完全不同：

不让每个线程都去竞争共享数据，而是选出一个线程当”组合者”（combiner），其他线程把操作请求发布到一个公共列表中。组合者批量执行所有请求，然后通知各线程。

线程 1: 请求 push(5) → 发布到请求列表
线程 2: 请求 pop()  → 发布到请求列表
线程 3: 请求 push(3) → 发布到请求列表

组合者（抢到锁的线程）：
  读取所有请求
  在私有栈上执行：push(5), pop(), push(3)
  写回结果给各线程

优势：组合者独占栈，没有 CAS 竞争；批量操作利用了 cache 局部性。

Flat Combining 在高竞争下性能极佳——甚至比 Elimination 更好，因为完全避免了 cache line 弹跳。

八、工程陷阱清单

九、总结与个人思考

Treiber 栈是并发编程的”原子”——最简单的无锁数据结构，但浓缩了所有核心问题。从它出发，可以清晰地看到并发数据结构的设计空间：

一、无锁不等于高性能。 Treiber 栈是无锁的，但在高竞争下比 mutex 好不了多少。根本原因是 LIFO 栈的语义要求串行化。无锁只是消除了锁的系统性问题（优先级反转、convoying），不能消除数据结构本身的串行瓶颈。

二、最好的优化是改变问题。 Elimination Stack 不是在”更快地做 CAS”——它改变了操作的路径：push 和 pop 直接交换数据，绕过了串行瓶颈。Flat Combining 更激进——把并发问题变回串行问题，但由一个线程高效地批量处理。

三、内存回收是真正的难题。 实现一个正确的 Treiber 栈不难。实现一个正确且不泄漏内存的 Treiber 栈才是真正的挑战。这也是为什么内存回收方案（Hazard Pointers、Epoch-Based、RCU）各自成为独立的研究领域。

四、Rust 的所有权系统在这里大放异彩。 Rust 的 crossbeam 库提供了 epoch::pin() API，编译器强制你在 pin 的作用域内才能访问共享指针。这把运行时的 use-after-free 风险变成了编译时错误。如果你在写并发数据结构，Rust 是目前最好的语言选择。

上一篇: 无锁队列：Michael-Scott 算法与 ABA 问题 下一篇: 并发跳表：ConcurrentSkipListMap 的设计

相关阅读: - Hazard Pointers - Epoch-Based Reclamation