Epoch-Based Reclamation：Crossbeam 的实现之道

// 线程 A 正在遍历链表，刚刚读取了指向节点 N 的指针
node_t *curr = atomic_load(&prev->next);  // curr 指向节点 N

// 与此同时，线程 B 通过 CAS 操作将节点 N 从链表中摘除
atomic_compare_exchange(&prev->next, &curr, curr->next);

// 问题来了：线程 B 现在能释放节点 N 的内存吗？
// 不能！因为线程 A 仍然持有指向 N 的指针
free(curr);  // 危险！线程 A 随时可能解引用这个指针

// Fraser 原始设计的核心数据结构（简化表示）
typedef struct {
    atomic_uint global_epoch;      // 全局纪元，只有 0, 1, 2 三个值

    struct thread_record {
        atomic_uint epoch;         // 线程本地纪元
        atomic_bool active;        // 是否在临界区中
        // 每个纪元一个垃圾袋
        garbage_bag_t limbo_list[3];
    } threads[MAX_THREADS];
} ebr_t;

void critical_enter(ebr_t *ebr, int tid) {
    atomic_store(&ebr->threads[tid].active, true);
    // 发布当前纪元——必须在 active=true 之后
    atomic_store(&ebr->threads[tid].epoch,
                 atomic_load(&ebr->global_epoch));
    atomic_thread_fence(memory_order_seq_cst);
}

void critical_exit(ebr_t *ebr, int tid) {
    atomic_thread_fence(memory_order_release);
    atomic_store(&ebr->threads[tid].active, false);
}

void try_gc(ebr_t *ebr) {
    uint32_t ge = atomic_load(&ebr->global_epoch);

    for (int i = 0; i < num_threads; i++) {
        if (atomic_load(&ebr->threads[i].active) &&
            atomic_load(&ebr->threads[i].epoch) != ge) {
            return;  // 有线程滞后，不能推进
        }
    }

    // 推进纪元，回收两个纪元前的垃圾
    uint32_t new_epoch = (ge + 1) % 3;
    if (atomic_compare_exchange_strong(&ebr->global_epoch, &ge, new_epoch)) {
        // 回收 new_epoch 对应的旧垃圾
        // (因为循环使用, garbage[new_epoch] 存的是三个纪元前的数据)
        free_garbage(new_epoch);
    }
}

// Crossbeam 中的 pin 用法
let guard = epoch::pin();  // 进入临界区，获取 Guard

// 在 guard 的保护下读取共享指针
let shared = some_atomic.load(Ordering::Acquire, &guard);

// shared 指向的内存在 guard 存活期间保证有效
if let Some(node) = unsafe { shared.as_ref() } {
    println!("value: {}", node.data);
}

// guard 被 drop 时自动 unpin
drop(guard);

// Guard 的简化定义
pub struct Guard {
    local: *const Local,  // 指向线程本地状态
}

impl Drop for Guard {
    fn drop(&mut self) {
        // 自动 unpin
        unsafe {
            (*self.local).unpin();
        }
    }
}

fn pin(&self) -> Guard {
    let guard_count = self.guard_count.get();
    self.guard_count.set(guard_count + 1);

    if guard_count == 0 {
        // 第一次 pin，需要真正进入临界区
        let global_epoch = self.global().epoch.load(Ordering::Relaxed);
        self.epoch.store(global_epoch, Ordering::Relaxed);
        atomic::fence(Ordering::SeqCst);
    }

    Guard { local: self }
}

fn unpin(&self) {
    let guard_count = self.guard_count.get();
    self.guard_count.set(guard_count - 1);

    if guard_count == 1 {
        // 最后一个 guard 被释放，真正离开临界区
        self.epoch.store(UNSET_EPOCH, Ordering::Release);

        if self.should_advance() {
            self.global().try_advance();
        }
    }
}

// 全局状态
pub struct Global {
    epoch: CachePadded<AtomicEpoch>,       // 全局纪元
    locals: List<Local>,                    // 所有线程本地状态的链表
    garbage: [CachePadded<Bag>; 3],         // 三个纪元的垃圾袋
}

// 线程本地状态
pub struct Local {
    epoch: AtomicEpoch,                     // 本地纪元
    guard_count: Cell<usize>,               // 嵌套 pin 计数
    bag: Cell<Bag>,                         // 本地垃圾袋
    collector: Collector,                   // 指向所属的 Collector
}

// 纪元值
struct Epoch {
    // 使用最低位作为 "pinned" 标志
    // 高位存储纪元值
    data: usize,
}

// 纪元编码：
// bit 0:      pinned 标志 (1 = pinned, 0 = unpinned)
// bit 1..63:  纪元值

const PINNED: usize = 1;

impl Epoch {
    fn is_pinned(&self) -> bool {
        self.data & PINNED != 0
    }

    fn unpinned() -> Self {
        Epoch { data: 0 }  // 纪元值无关紧要，因为未 pin
    }

    fn pinned(epoch: usize) -> Self {
        Epoch { data: (epoch << 1) | PINNED }
    }

    fn value(&self) -> usize {
        self.data >> 1
    }
}

// Collector 是全局的，可以 Clone（引用计数）
let collector = Collector::new();

// 每个线程通过 register 获取 Handle
let handle = collector.register();

// 通过 Handle 进行 pin
let guard = handle.pin();

// 退休一个节点
unsafe fn defer_destroy<T>(guard: &Guard, ptr: Shared<T>) {
    let local = guard.local();
    let bag = local.bag.get_mut();

    bag.push(Deferred::new(move || {
        drop(Box::from_raw(ptr.as_raw() as *mut T));
    }));

    if bag.is_full() {
        let bag = local.bag.replace(Bag::new());
        let epoch = local.epoch.load(Ordering::Relaxed);
        local.global().push_bag(bag, epoch);
    }
}

fn try_advance(&self) -> bool {
    let global_epoch = self.epoch.load(Ordering::Relaxed);

    // 检查所有注册的线程
    for local in self.locals.iter() {
        let local_epoch = local.epoch.load(Ordering::Relaxed);

        // 如果线程已 pin 且纪元落后，不能推进
        if local_epoch.is_pinned()
            && local_epoch.value() != global_epoch
        {
            return false;
        }
    }

    // 所有线程都跟上了，推进纪元
    atomic::fence(Ordering::Acquire);
    let new_epoch = global_epoch.wrapping_add(1);
    self.epoch.store(new_epoch, Ordering::Release);

    // 回收两个纪元前的垃圾
    let old_epoch = new_epoch.wrapping_add(1);  // (e+2) % 3 == (e-1) % 3
    self.garbage[old_epoch % 3].collect();

    true
}

#[repr(align(128))]  // 通常缓存行是 64 字节，128 保证跨越行边界
pub struct CachePadded<T> {
    value: T,
}

impl Guard {
    // 返回的 Shared 的生命周期被绑定到 &self（即 Guard 的生命周期）
    pub fn load<'g, T>(
        &'g self,
        atomic: &Atomic<T>,
        ord: Ordering,
    ) -> Shared<'g, T> {
        // Shared<'g, T> 不能比 Guard 活得更久
        Shared {
            ptr: atomic.load(ord),
            _marker: PhantomData,
        }
    }
}

// 编译器会阻止这样的错误用法：
let shared;
{
    let guard = epoch::pin();
    shared = some_atomic.load(Ordering::Acquire, &guard);
}  // guard 被 drop，unpin
// shared.as_ref();  // 编译错误！shared 的生命周期超过了 guard

// Shared<'g, T> - 受 Guard 保护的共享引用
// 类似于 &'g T，但来自原子指针
pub struct Shared<'g, T: 'g> {
    ptr: *const T,
    _marker: PhantomData<(&'g (), *const T)>,
}

// Owned<T> - 独占所有权，可以安全 drop
// 类似于 Box<T>
pub struct Owned<T> {
    ptr: *mut T,
    _marker: PhantomData<Box<T>>,
}

// Atomic<T> - 原子指针，类似于 AtomicPtr<T>
pub struct Atomic<T> {
    ptr: AtomicUsize,
    _marker: PhantomData<*mut T>,
}

/* epoch_reclaim.h - 简化的 Epoch-Based Reclamation 实现 */
#ifndef EPOCH_RECLAIM_H
#define EPOCH_RECLAIM_H

#include <stdatomic.h>
#include <stdbool.h>
#include <stdlib.h>
#include <string.h>

/* ===== 配置参数 ===== */
#define EBR_MAX_THREADS     64
#define EBR_EPOCHS          3
#define EBR_BAG_CAPACITY    128
#define EBR_ADVANCE_FREQ    64    /* 每隔多少次 unpin 尝试一次推进 */

/* ===== 待回收节点 ===== */
typedef void (*ebr_free_fn)(void *ptr);

typedef struct ebr_retired_node {
    void               *ptr;
    ebr_free_fn         free_fn;
    struct ebr_retired_node *next;
} ebr_retired_node_t;

/* ===== 线程本地垃圾袋 ===== */
typedef struct {
    ebr_retired_node_t *head;
    int                 count;
} ebr_bag_t;

/* ===== 线程本地记录 ===== */
typedef struct {
    _Atomic(unsigned)   epoch;          /* 线程观察到的纪元 */
    _Atomic(bool)       active;         /* 是否在临界区中 */
    _Atomic(bool)       used;           /* 是否已注册 */
    unsigned            pin_count;      /* 嵌套 pin 计数 */
    unsigned            op_count;       /* 操作计数，用于触发推进 */
    ebr_bag_t           bags[EBR_EPOCHS]; /* 每个纪元一个垃圾袋 */
    char                _pad[64];       /* 缓存行填充 */
} ebr_thread_t;

/* ===== 全局 EBR 状态 ===== */
typedef struct {
    _Atomic(unsigned)   global_epoch;
    ebr_thread_t        threads[EBR_MAX_THREADS];
    int                 max_tid;        /* 记录最大注册线程 ID */
} ebr_t;

/* ===== 初始化 ===== */
static inline void ebr_init(ebr_t *ebr)
{
    memset(ebr, 0, sizeof(ebr_t));
    atomic_store(&ebr->global_epoch, 0);
    ebr->max_tid = 0;
}

/* ===== 线程注册 ===== */
static inline int ebr_register(ebr_t *ebr)
{
    for (int i = 0; i < EBR_MAX_THREADS; i++) {
        bool expected = false;
        if (atomic_compare_exchange_strong(&ebr->threads[i].used,
                                           &expected, true)) {
            atomic_store(&ebr->threads[i].active, false);
            atomic_store(&ebr->threads[i].epoch, 0);
            ebr->threads[i].pin_count = 0;
            ebr->threads[i].op_count = 0;
            for (int e = 0; e < EBR_EPOCHS; e++) {
                ebr->threads[i].bags[e].head = NULL;
                ebr->threads[i].bags[e].count = 0;
            }
            /* 更新 max_tid（非精确，但足够） */
            int cur_max = ebr->max_tid;
            while (i >= cur_max) {
                __atomic_compare_exchange_n(&ebr->max_tid,
                    &cur_max, i + 1, true,
                    __ATOMIC_RELAXED, __ATOMIC_RELAXED);
            }
            return i;
        }
    }
    return -1;  /* 注册失败，超出最大线程数 */
}

/* ===== 线程注销 ===== */
static inline void ebr_unregister(ebr_t *ebr, int tid)
{
    atomic_store(&ebr->threads[tid].active, false);
    /* 注意：注销时垃圾袋中可能还有待回收节点
       完整实现需要将这些节点移交给全局垃圾队列 */
    for (int e = 0; e < EBR_EPOCHS; e++) {
        ebr_retired_node_t *node = ebr->threads[tid].bags[e].head;
        while (node) {
            ebr_retired_node_t *next = node->next;
            node->free_fn(node->ptr);
            free(node);
            node = next;
        }
        ebr->threads[tid].bags[e].head = NULL;
        ebr->threads[tid].bags[e].count = 0;
    }
    atomic_store(&ebr->threads[tid].used, false);
}

/* ===== 回收指定纪元的垃圾 ===== */
static inline void ebr_collect_epoch(ebr_t *ebr, unsigned epoch_idx)
{
    int max = ebr->max_tid;
    for (int i = 0; i < max; i++) {
        if (!atomic_load(&ebr->threads[i].used)) continue;

        ebr_bag_t *bag = &ebr->threads[i].bags[epoch_idx];
        ebr_retired_node_t *node = bag->head;
        while (node) {
            ebr_retired_node_t *next = node->next;
            node->free_fn(node->ptr);
            free(node);
            node = next;
        }
        bag->head = NULL;
        bag->count = 0;
    }
}

/* ===== 尝试推进全局纪元 ===== */
static inline bool ebr_try_advance(ebr_t *ebr)
{
    unsigned ge = atomic_load_explicit(&ebr->global_epoch,
                                       memory_order_relaxed);
    int max = ebr->max_tid;

    /* 检查所有活跃线程是否已跟上当前纪元 */
    for (int i = 0; i < max; i++) {
        if (!atomic_load_explicit(&ebr->threads[i].used,
                                   memory_order_relaxed))
            continue;

        if (atomic_load_explicit(&ebr->threads[i].active,
                                  memory_order_acquire)) {
            unsigned te = atomic_load_explicit(&ebr->threads[i].epoch,
                                                memory_order_relaxed);
            if (te != ge) {
                return false;  /* 有线程滞后 */
            }
        }
    }

    /* 所有线程已跟上，推进纪元 */
    unsigned new_epoch = (ge + 1) % EBR_EPOCHS;

    /* CAS 避免多线程同时推进 */
    if (!atomic_compare_exchange_strong_explicit(
            &ebr->global_epoch, &ge, new_epoch,
            memory_order_release, memory_order_relaxed)) {
        return false;  /* 其他线程已推进 */
    }

    /* 回收新纪元槽位上的旧垃圾
       (new_epoch 槽位存放的是三个纪元前的垃圾) */
    ebr_collect_epoch(ebr, new_epoch);

    return true;
}

/* ===== Pin：进入临界区 ===== */
static inline void ebr_pin(ebr_t *ebr, int tid)
{
    ebr_thread_t *t = &ebr->threads[tid];

    if (t->pin_count++ > 0) {
        return;  /* 嵌套 pin，只增加计数 */
    }

    /* 标记为活跃 */
    atomic_store_explicit(&t->active, true, memory_order_relaxed);

    /* 加载并发布当前全局纪元 */
    unsigned ge = atomic_load_explicit(&ebr->global_epoch,
                                       memory_order_relaxed);
    atomic_store_explicit(&t->epoch, ge, memory_order_relaxed);

    /* 关键屏障：确保纪元更新在后续所有读操作之前可见 */
    atomic_thread_fence(memory_order_seq_cst);
}

/* ===== Unpin：离开临界区 ===== */
static inline void ebr_unpin(ebr_t *ebr, int tid)
{
    ebr_thread_t *t = &ebr->threads[tid];

    if (--t->pin_count > 0) {
        return;  /* 嵌套 unpin，只减少计数 */
    }

    /* 释放屏障：确保临界区内的操作完成 */
    atomic_thread_fence(memory_order_release);
    atomic_store_explicit(&t->active, false, memory_order_relaxed);

    /* 周期性尝试推进纪元 */
    if (++t->op_count % EBR_ADVANCE_FREQ == 0) {
        ebr_try_advance(ebr);
    }
}

/* ===== 退休节点：将节点放入当前纪元的垃圾袋 ===== */
static inline void ebr_retire(ebr_t *ebr, int tid,
                               void *ptr, ebr_free_fn free_fn)
{
    unsigned ge = atomic_load_explicit(&ebr->global_epoch,
                                       memory_order_relaxed);
    unsigned idx = ge % EBR_EPOCHS;

    ebr_retired_node_t *node = (ebr_retired_node_t *)malloc(
        sizeof(ebr_retired_node_t));
    node->ptr = ptr;
    node->free_fn = free_fn;
    node->next = ebr->threads[tid].bags[idx].head;
    ebr->threads[tid].bags[idx].head = node;
    ebr->threads[tid].bags[idx].count++;
}

/* ===== 强制回收（用于 shutdown） ===== */
static inline void ebr_drain(ebr_t *ebr)
{
    for (unsigned e = 0; e < EBR_EPOCHS; e++) {
        ebr_collect_epoch(ebr, e);
    }
}

#endif /* EPOCH_RECLAIM_H */

// 危险！不要这样做
let guard = epoch::pin();

// 长时间阻塞操作
std::thread::sleep(Duration::from_secs(60));

// 在这 60 秒内，全局纪元无法推进
// 所有线程的垃圾都在积累
drop(guard);

ebr_pin(&ebr, tid);
/* 线程被 OS 挂起，或者 */
while (some_lock_free_cas_keeps_failing) {
    /* CAS 重试循环——pin 期间不断重试 */
}
ebr_unpin(&ebr, tid);

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

pub struct TreiberStack<T> {
    head: Atomic<Node<T>>,
}

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

impl<T> TreiberStack<T> {
    pub fn new() -> Self {
        TreiberStack {
            head: Atomic::null(),
        }
    }

    /// 压栈操作
    pub fn push(&self, data: T) {
        let node = Owned::new(Node {
            data,
            next: Atomic::null(),
        });

        let guard = epoch::pin();

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

            match self.head.compare_exchange(
                head,
                node,
                Ordering::Release,
                Ordering::Relaxed,
                &guard,
            ) {
                Ok(_) => break,
                Err(e) => {
                    // CAS 失败，node 通过 e.new 返回
                    // 编译器帮助我们避免了内存泄漏
                    node = e.new;
                }
            }
        }
    }

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

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

            let head_ref = unsafe {
                match head.as_ref() {
                    Some(r) => r,
                    None => return None,  // 栈空
                }
            };

            let next = head_ref.next.load(Ordering::Acquire, &guard);

            if self.head.compare_exchange(
                head,
                next,
                Ordering::Release,
                Ordering::Relaxed,
                &guard,
            ).is_ok() {
                // CAS 成功，读取数据并延迟释放节点
                let data = unsafe { ptr::read(&head_ref.data) };
                unsafe {
                    guard.defer_destroy(head);
                }
                return Some(data);
            }
            // CAS 失败，重试
        }
    }
}

impl<T> Drop for TreiberStack<T> {
    fn drop(&mut self) {
        while self.pop().is_some() {}
    }
}

/* treiber_stack.c - 基于 EBR 的 Treiber 栈 */
#include "epoch_reclaim.h"
#include <stdatomic.h>
#include <stdio.h>
#include <pthread.h>

/* ===== 栈节点 ===== */
typedef struct stack_node {
    int                     value;
    _Atomic(struct stack_node *) next;
} stack_node_t;

/* ===== Treiber 栈 ===== */
typedef struct {
    _Atomic(stack_node_t *) top;
    ebr_t                   ebr;
} treiber_stack_t;

static void free_stack_node(void *ptr) {
    free(ptr);
}

void stack_init(treiber_stack_t *s) {
    atomic_store(&s->top, NULL);
    ebr_init(&s->ebr);
}

void stack_push(treiber_stack_t *s, int tid, int value) {
    stack_node_t *node = (stack_node_t *)malloc(sizeof(stack_node_t));
    node->value = value;

    ebr_pin(&s->ebr, tid);

    stack_node_t *old_top;
    do {
        old_top = atomic_load_explicit(&s->top, memory_order_acquire);
        atomic_store_explicit(&node->next, old_top, memory_order_relaxed);
    } while (!atomic_compare_exchange_weak_explicit(
        &s->top, &old_top, node,
        memory_order_release, memory_order_relaxed));

    ebr_unpin(&s->ebr, tid);
}

bool stack_pop(treiber_stack_t *s, int tid, int *out) {
    ebr_pin(&s->ebr, tid);

    stack_node_t *old_top;
    stack_node_t *new_top;

    do {
        old_top = atomic_load_explicit(&s->top, memory_order_acquire);
        if (old_top == NULL) {
            ebr_unpin(&s->ebr, tid);
            return false;
        }
        new_top = atomic_load_explicit(&old_top->next,
                                        memory_order_relaxed);
    } while (!atomic_compare_exchange_weak_explicit(
        &s->top, &old_top, new_top,
        memory_order_release, memory_order_relaxed));

    *out = old_top->value;

    /* 延迟释放：将旧栈顶节点放入当前纪元的垃圾袋 */
    ebr_retire(&s->ebr, tid, old_top, free_stack_node);

    ebr_unpin(&s->ebr, tid);
    return true;
}

/* ===== 测试 ===== */
#define NUM_THREADS 4
#define OPS_PER_THREAD 10000

treiber_stack_t g_stack;
_Atomic(int) g_push_sum = 0;
_Atomic(int) g_pop_sum = 0;

void *push_worker(void *arg) {
    int tid = ebr_register(&g_stack.ebr);
    int local_sum = 0;

    for (int i = 1; i <= OPS_PER_THREAD; i++) {
        stack_push(&g_stack, tid, i);
        local_sum += i;
    }

    atomic_fetch_add(&g_push_sum, local_sum);
    ebr_unregister(&g_stack.ebr, tid);
    return NULL;
}

void *pop_worker(void *arg) {
    int tid = ebr_register(&g_stack.ebr);
    int local_sum = 0;
    int value;
    int empty_count = 0;

    while (empty_count < 1000) {
        if (stack_pop(&g_stack, tid, &value)) {
            local_sum += value;
            empty_count = 0;
        } else {
            empty_count++;
            sched_yield();
        }
    }

    atomic_fetch_add(&g_pop_sum, local_sum);
    ebr_unregister(&g_stack.ebr, tid);
    return NULL;
}

int main(void) {
    stack_init(&g_stack);

    pthread_t pushers[NUM_THREADS], poppers[NUM_THREADS];

    for (int i = 0; i < NUM_THREADS; i++)
        pthread_create(&pushers[i], NULL, push_worker, NULL);
    for (int i = 0; i < NUM_THREADS; i++)
        pthread_create(&poppers[i], NULL, pop_worker, NULL);
    for (int i = 0; i < NUM_THREADS; i++)
        pthread_join(pushers[i], NULL);
    for (int i = 0; i < NUM_THREADS; i++)
        pthread_join(poppers[i], NULL);

    /* 弹出剩余元素并验证 */
    int tid = ebr_register(&g_stack.ebr);
    int value;
    while (stack_pop(&g_stack, tid, &value))
        atomic_fetch_add(&g_pop_sum, value);
    ebr_unregister(&g_stack.ebr, tid);
    ebr_drain(&g_stack.ebr);

    printf("pushed=%d popped=%d match=%s\n",
           atomic_load(&g_push_sum), atomic_load(&g_pop_sum),
           atomic_load(&g_push_sum) == atomic_load(&g_pop_sum) ? "YES" : "NO");
    return 0;
}