上次我们试图实现一个无锁的并发栈，但是发现由于Rust没有GC，简单的实现会导致内存泄漏。于是crossbeam提供了一个基于epoch的“垃圾收集”(epoch based reclamation)库。首先来简单的说一下这一垃圾回收的原理。

基于epoch的“垃圾回收”

这一算法主要在Keir Fraser的博士论文中有涉及。我们从之前无锁并发栈的简单实现中看到，并发环境中的数据结构的容器（比如包含了出栈元素的那个节点），会在一个线程中完成它的使命然后准备被释放。但是这个时候可能还有其他线程还持有其快照（主要是为了进行CAS）。不过不会再有新的线程还会继续取得其快照了。那么只要等到所有对该容器取得快照的线程都完成操作，就可以安心的释放这个容器了。

这一算法需要：

1. 一个全局的epoch计数器
2. 每个线程有一个本地的epoch计数器
3. 一个全局列表记录每个epoch产生的垃圾
4. 标示线程是否活跃的标示符

算法大概是这样的：当一个线程需要对某个数据结构操作时，它先将自己的标识符设置为活跃，然后将本地epoch计数器更新为全局epoch计数器。而如果线程需要从数据结构中删除一个节点，它会把该节点加入当前epoch的垃圾列表。当线程完成对数据结构的操作时，线程将自己标记为不活跃。然后当需要收集垃圾的时候，一个线程会遍历所有线程并检查是否所有活跃的线程的epoch都和当前全局的epoch相同，如果是那么全局epoch计数将会+1，同时2个epoch前的垃圾可以被回收。

事实上，当处于epoch N+2的时候，由于所有对数据进行操作（活跃）的线程都在epoch N+1之后，可以安全的清理epoch N的垃圾列表。

crossbeam-epoch的API

Guard

使用pin可以产生Guard，这将当前线程标识为活跃

use crossbeam_epoch as epoch;

let guard = &epoch::pin();

defer_destroy则将数据放入垃圾列表，等待2个epoch之后就可以被清理了

pub unsafe fn defer_destroy<T>(&self, ptr: Shared<T>)

然后是我们将看到三大指针类型：

Shared

我们刚刚在defer_destroy的函数签名中看到了Shared<T>，这是一个被Guard的生命周期'g所保护的指针类型，相当于&'g T。它保证了Guard存在期间数据的可访问。

Owned

这相当于Box<T>，是一个不会被其他线程接触到的量。

Atomic

这是一个可以在线程间交换的原子指针。可以在Guard的保护下从中读出Shared<T>

pub fn load(&self, ord: Ordering, &'g Guard) -> Shared<'g, T>

也可以将Owned<T>或者Shared<T>（即Pointer<T>）存入

pub fn store<P>(&self, new: P, ord: Ordering)
where
    P: Pointer<T>, 

也可以对其进行CAS操作

pub fn compare_and_set<O, P>(
    &self,
    current: Shared<T>,
    new: P,
    ord: O,
    &'g Guard
) -> Result<Shared<'g, T>, CompareAndSetError<'g, T, P>>
where
    O: CompareAndSetOrdering,
    P: Pointer<T>, 

实现并发栈

有了这些API，我们就可以实现一个无锁的MPMC的Treiber栈了。

use std::mem::ManuallyDrop;
use std::ptr;
use std::sync::atomic::Ordering::{Acquire, Relaxed, Release};

use crossbeam_epoch::{self as epoch, Atomic, Owned};

#[derive(Debug)]
pub struct TreiberStack<T> {
    head: Atomic<Node<T>>,
}

#[derive(Debug)]
struct Node<T> {
    data: ManuallyDrop<T>, // 告诉编译器该变量不需要自动Drop
    next: Atomic<Node<T>>,
}

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

    pub fn push(&self, t: T) {
        let mut n = Owned::new(Node {
            data: ManuallyDrop::new(t),
            next: Atomic::null(),
        });

        let guard = epoch::pin(); // 标记当前线程为活跃

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

            match self.head.compare_and_set(head, n, Release, &guard) { // CAS
                Ok(_) => break,
                Err(e) => n = e.new,
            }
        }
    }

    pub fn pop(&self) -> Option<T> {
        let guard = epoch::pin(); // 标记当前线程为活跃
        loop {
            let head = self.head.load(Acquire, &guard);

            match unsafe { head.as_ref() } {
                Some(h) => {
                    let next = h.next.load(Relaxed, &guard);

                    if self
                        .head
                        .compare_and_set(head, next, Relaxed, &guard) // CAS
                        .is_ok()
                    {
                        unsafe {
                            guard.defer_destroy(head); // 将垃圾加入列表
                            return Some(ManuallyDrop::into_inner(ptr::read(&(*h).data))); // 返回节点中的数据
                        }
                    }
                }
                None => return None,
            }
        }
    }

    pub fn is_empty(&self) -> bool {
        let guard = epoch::pin();
        self.head.load(Acquire, &guard).is_null()
    }
}

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

小结

很好，该系列的第一个库、第一个并发算法和第一个并发数据结构已经讲完了。是不是觉得epoch的API很简洁好用呢？下一次我们会将结合crossbeam-deque，讲一下work-stealing算法，敬请期待。