4. 无锁栈（Lock-Free Stack）：Treiber Stack实现、内存管理（RCU与Hazard Pointer）

无锁栈（Lock-Free Stack）是多线程编程中实现高性能并发的基础数据结构之一。本章深入剖析经典的 Treiber Stack 算法，并重点解决无锁编程中最棘手的问题——内存管理，对比 RCU（Read-Copy-Update） 与 Hazard Pointer 两种主流方案在 DPDK 环境下的工程选型。

4.1 Treiber Stack 算法原理

Treiber Stack 由 R. Kent Treiber 于 1986 年提出，是基于 CAS（Compare-And-Swap） 操作实现的并发无锁栈。其核心思想是：通过原子操作更新栈顶指针，避免传统锁带来的上下文切换开销。

4.1.1 数据结构定义

// Treiber Stack 节点结构
struct treiber_node {
    void *data;                 // 数据指针
    struct treiber_node *next;  // 下一节点
};

// Treiber Stack 栈顶指针
struct treiber_stack {
    struct treiber_node *top;   // 原子操作栈顶
};

4.1.2 Push 操作流程

1. 创建新节点：分配内存并填充数据。
2. 读取当前栈顶：获取 top 指针。
3. 链接新节点：将新节点的 next 指向当前栈顶。
4. CAS 更新栈顶：原子比较 top 是否仍为之前读取的值，若是则更新为新节点；否则重试步骤 2-4。

void treiber_push(struct treiber_stack *stack, void *data) {
    struct treiber_node *node = malloc(sizeof(struct treiber_node));
    node->data = data;

    struct treiber_node *old_top;
    do {
        old_top = atomic_load(&stack->top);
        node->next = old_top;
    } while (!atomic_compare_exchange_weak(&stack->top, &old_top, node));
}

4.1.3 Pop 操作流程

1. 读取当前栈顶：获取 top 指针。
2. 检查空栈：若 top == NULL，返回空。
3. CAS 更新栈顶：原子比较 top 是否仍为之前读取的值，若是则更新为 top->next；否则重试步骤 1-3。
4. 返回数据：成功弹出后返回节点数据。

void *treiber_pop(struct treiber_stack *stack) {
    struct treiber_node *old_top;
    do {
        old_top = atomic_load(&stack->top);
        if (old_top == NULL) return NULL;  // 空栈
    } while (!atomic_compare_exchange_weak(&stack->top, &old_top, old_top->next));

    void *data = old_top->data;
    // 注意：此处不能直接 free(old_top)，存在 ABA 问题
    return data;
}

4.2 Treiber Stack 的 ABA 问题与解决方案

ABA 问题是无锁数据结构中常见的陷阱。解决思路主要有两种：标记指针（Tagged Pointer） 和 内存回收延迟（Safe Memory Reclamation, SMR）。

4.3 内存管理方案一：Hazard Pointer

Hazard Pointer（危险指针）由 Maged M. Michael 于 2004 年提出。每个线程维护一个 Hazard Pointer 数组，声明当前正在访问的共享指针。其他线程在释放内存前，需检查该指针是否被任何线程声明为危险。

4.3.1 核心数据结构

#define MAX_THREADS 128
#define HP_MAX 4  // 每个线程最多同时声明的危险指针数

struct hp_thread {
    void *hazard_ptrs[HP_MAX];  // 危险指针数组
    int hp_count;               // 当前声明数量
};

struct hp_retire_list {
    void *ptr;
    struct hp_retire_list *next;
};

// 全局退役列表（每个线程一个）
__thread struct hp_retire_list *retire_list = NULL;

4.3.2 操作流程

1. 声明危险指针：读取共享指针前，将其写入线程的 Hazard Pointer 数组。
2. 验证指针有效性：再次读取共享指针，确认与声明的一致（防止指针已被修改）。
3. 使用指针：安全访问数据。
4. 解除声明：使用完毕后，清除 Hazard Pointer。
5. 延迟回收：当节点需要释放时，先放入退役列表，定期扫描所有线程的 Hazard Pointer，确认无引用后再释放。

// 声明危险指针
void hp_protect(struct hp_thread *thr, int index, void *ptr) {
    thr->hazard_ptrs[index] = ptr;
    atomic_thread_fence(memory_order_seq_cst);
}

// 退役节点（延迟释放）
void hp_retire(struct hp_thread *thr, void *ptr) {
    struct hp_retire_list *node = malloc(sizeof(struct hp_retire_list));
    node->ptr = ptr;
    node->next = retire_list;
    retire_list = node;

    // 当退役列表超过阈值时，触发扫描回收
    if (++thr->retire_count >= HP_RETIRE_THRESHOLD) {
        hp_scan(thr);
    }
}

4.4 内存管理方案二：RCU（Read-Copy-Update）

RCU 是 Linux 内核中广泛使用的同步机制，特别适合 读多写少 的场景。读者无需任何原子操作或内存屏障（在特定架构下），写者通过 宽限期（Grace Period） 确保所有读者完成对旧数据的访问后再回收内存。

4.4.1 RCU 核心概念

4.4.2 RCU 在 Treiber Stack 中的应用

// RCU 保护的 Treiber Stack Pop
void *rcu_treiber_pop(struct treiber_stack *stack) {
    struct treiber_node *old_top;

    rcu_read_lock();  // 进入 RCU 读临界区
    do {
        old_top = rcu_dereference(stack->top);
        if (old_top == NULL) {
            rcu_read_unlock();
            return NULL;
        }
    } while (!atomic_compare_exchange_weak(&stack->top, &old_top,
                                           rcu_dereference(old_top->next)));
    rcu_read_unlock();

    void *data = old_top->data;
    // 宽限期后安全释放
    call_rcu(&old_top->rcu_head, rcu_node_free_callback);
    return data;
}

4.5 RCU vs Hazard Pointer：工程选型对比

在 DPDK 高性能网络处理环境中，选择合适的内存管理方案至关重要。以下从多个维度进行对比：

4.6 无锁栈性能对比与选型建议

以下是在 DPDK 环境下，不同实现方案的性能对比（基于 8 核 Xeon 处理器，1000 万次操作）：

4.7 本章小结

• Treiber Stack 通过 CAS 循环实现无锁入栈和出栈，但存在 ABA 问题。
• ABA 问题 的解决方案包括标记指针、Hazard Pointer 和 RCU。
• Hazard Pointer 通过线程声明危险指针实现安全内存回收，适合读写均衡场景。
• RCU 通过宽限期机制实现读者无锁访问，适合读多写少场景，DPDK 提供原生支持。
• 工程选型需综合考虑读者/写者开销、内存占用、实现复杂度和 DPDK 生态兼容性。