crossbeam-epoch（三）：侵入式无锁链表

crossbeam-epoch 中使用侵入式无锁链表来组织所有线程的 Local：https://github.com/crossbeam-rs/crossbeam/blob/master/crossbeam-epoch/src/sync/list.rs

侵入式链表

侵入式链表的内存布局如下：

Node 是业务数据节点，由使用者来定义，Node 中必须包含一个 entry，每个 entry 中都保存指向下一个节点的 entry 的指针，这样才能把所有的节点串联起来构成一个单链表。

显而易见，侵入式链表与非侵入式链表主要有两个不同点：

• 侵入式链表的节点 Node 由使用者来定义，而非侵入式链表由实现者来定义；
• 侵入式链表的节点 Node 中的字段 entry 保存指向下一个节点的 entry 的指针，以此来串联所有的节点。而非侵入式链表中通过 next 指针指向下一个 Node 节点来串联形成链表。

例如，一个典型的非侵入式链表的内存布局如下所示：

使用侵入式链表的一个显而易见的好处是不需要维护泛型参数，降低实现的复杂度。由于侵入式链表的节点由使用者定义，那么就可以在节点中任意增加字段：

struct Node {
	entry: Entry,
	
	filed1: i32,
	filed2: String,
	filed3: usize,
	.....
	.....
}

而非侵入式链表则不行，链表的结点由实现者定义，实现者不可能预先知道使用者需要哪些类型的字段，因此只能使用泛型参数，而使用者就必须把需要的类型打包成一个单独的类型，然后才能使用链表：

struct Node<T> {
	data: T,
	next: *mut Node,
}

crossbeam-epoch 中非侵入链表的定义如下：

/// An entry in a linked list.
///
/// An Entry is accessed from multiple threads, so it would be beneficial to put it in a different
/// cache-line than thread-local data in terms of performance.
#[derive(Debug)]
pub(crate) struct Entry {
    /// The next entry in the linked list.
    /// If the tag is 1, this entry is marked as deleted.
    next: Atomic<Entry>,
}

/// A lock-free, intrusive linked list of type `T`.
#[derive(Debug)]
pub(crate) struct List<T, C: IsElement<T> = T> {
    /// The head of the linked list.
    head: Atomic<Entry>,

    /// The phantom data for using `T` and `C`.
    _marker: PhantomData<(T, C)>,
}

IsElement trait

在侵入式链表中，可以通过节点中的 entry 找到下一个节点的 entry，但是要如何通过 entry 访问完整的节点呢？假设节点的定义如下：

#[repr(c)]
struct Node {
	counter: usize,
	entry: Entry,
	key: i32,
	value: i32,
}

内存布局：

假设存在引用 &Entry，那么可以通过减去 entry 在 Node 中的地址偏移量得到 Node 的指针：

(&Entry as usize - offset_of!(A, entry)) as *const Node

同样的，如果存在引用 &Node，也可以通过加上 entry 在 Node 中的地址偏移量得到 Entry 指针：

(&Node as usize + offset_of!(A, entry)) as *const Entry

为了统一上述的 entry 和 Node 转换操作，crossbeam-epoch 定义了 IsElement<T> trait：

/// Implementing this trait asserts that the type `T` can be used as an element in the intrusive
/// linked list defined in this module. `T` has to contain (or otherwise be linked to) an instance
/// of `Entry`.
pub(crate) trait IsElement<T> {
    /// Returns a reference to this element's `Entry`.
    fn entry_of(_: &T) -> &Entry;

    /// Given a reference to an element's entry, returns that element.
    ///
    /// ```ignore
    /// let elem = ListElement::new();
    /// assert_eq!(elem.entry_of(),
    ///            unsafe { ListElement::element_of(elem.entry_of()) } );
    /// ```
    ///
    /// # Safety
    ///
    /// The caller has to guarantee that the `Entry` is called with was retrieved from an instance
    /// of the element type (`T`).
    unsafe fn element_of(_: &Entry) -> &T;

    /// The function that is called when an entry is unlinked from list.
    ///
    /// # Safety
    ///
    /// The caller has to guarantee that the `Entry` is called with was retrieved from an instance
    /// of the element type (`T`).
    unsafe fn finalize(_: &Entry, _: &Guard);
}

entry_of 和 element_of 定义了侵入式链表中 &Node 和 &Entry 之间的互相转换接口。finalize 定义了回收链表中节点的接口。

使用示例：

struct A {
	entry: Entry,
    data: usize,
}

impl IsElement<A> for A {
    fn entry_of(a: &A) -> &Entry {
        let entry_ptr = ((a as usize) + offset_of!(A, entry)) as *const Entry;
        unsafe { &*entry_ptr }
    }

    unsafe fn element_of(entry: &Entry) -> &T {
        let elem_ptr = ((entry as usize) - offset_of!(A, entry)) as *const T;
        &*elem_ptr
    }

    unsafe fn finalize(entry: &Entry, guard: &Guard) {
        guard.defer_destroy(Shared::from(Self::element_of(entry) as *const _));
    }
}

insert

插入一个新节点到侵入式链表的头部：

/// Inserts `entry` into the head of the list.
    ///
    /// # Safety
    ///
    /// You should guarantee that:
    ///
    /// - `container` is not null
    /// - `container` is immovable, e.g. inside an `Owned`
    /// - the same `Entry` is not inserted more than once
    /// - the inserted object will be removed before the list is dropped
    pub(crate) unsafe fn insert<'g>(&'g self, container: Shared<'g, T>, guard: &'g Guard) {
        // Insert right after head, i.e. at the beginning of the list.
        let to = &self.head;
        // Get the intrusively stored Entry of the new element to insert.
        let entry: &Entry = C::entry_of(container.deref());
        // Make a Shared ptr to that Entry.
        let entry_ptr = Shared::from(entry as *const _);
        // Read the current successor of where we want to insert.
        let mut next = to.load(Relaxed, guard);

        loop {
            // Set the Entry of the to-be-inserted element to point to the previous successor of
            // `to`.
            entry.next.store(next, Relaxed);
            match to.compare_exchange_weak(next, entry_ptr, Release, Relaxed, guard) {
                Ok(_) => break,
                // We lost the race or weak CAS failed spuriously. Update the successor and try
                // again.
                Err(err) => next = err.current,
            }
        }
    }

插入逻辑如下图所示：

delete

删除节点：

impl Entry {
    /// Marks this entry as deleted, deferring the actual deallocation to a later iteration.
    ///
    /// # Safety
    ///
    /// The entry should be a member of a linked list, and it should not have been deleted.
    /// It should be safe to call `C::finalize` on the entry after the `guard` is dropped, where `C`
    /// is the associated helper for the linked list.
    pub(crate) unsafe fn delete(&self, guard: &Guard) {
        self.next.fetch_or(1, Release, guard);
    }
}

删除节点仅仅是把 entry 的 next 原子指针的 tag 设置为1，标记这个节点已经在逻辑上被删除了。之所以不立即把这个节点从链表中删除，是为了避免由于并发的插入/删除节点导致插入成功的节点被丢失的问题。

traverse

首先创建迭代器：

/// Returns an iterator over all objects.
    ///
    /// # Caveat
    ///
    /// Every object that is inserted at the moment this function is called and persists at least
    /// until the end of iteration will be returned. Since this iterator traverses a lock-free
    /// linked list that may be concurrently modified, some additional caveats apply:
    ///
    /// 1. If a new object is inserted during iteration, it may or may not be returned.
    /// 2. If an object is deleted during iteration, it may or may not be returned.
    /// 3. The iteration may be aborted when it lost in a race condition. In this case, the winning
    ///    thread will continue to iterate over the same list.
    pub(crate) fn iter<'g>(&'g self, guard: &'g Guard) -> Iter<'g, T, C> {
        Iter {
            guard,
            pred: &self.head,
            curr: self.head.load(Acquire, guard),
            head: &self.head,
            _marker: PhantomData,
        }
    }

遍历迭代器：

impl<'g, T: 'g, C: IsElement<T>> Iterator for Iter<'g, T, C> {
    type Item = Result<&'g T, IterError>;

    fn next(&mut self) -> Option<Self::Item> {
        while let Some(c) = unsafe { self.curr.as_ref() } {
            let succ = c.next.load(Acquire, self.guard);

            if succ.tag() == 1 {
                // This entry was removed. Try unlinking it from the list.
                let succ = succ.with_tag(0);

                // The tag should always be zero, because removing a node after a logically deleted
                // node leaves the list in an invalid state.
                debug_assert!(self.curr.tag() == 0);

                // Try to unlink `curr` from the list, and get the new value of `self.pred`.
                let succ = match self
                    .pred
                    .compare_exchange(self.curr, succ, Acquire, Acquire, self.guard)
                {
                    Ok(_) => {
                        // We succeeded in unlinking `curr`, so we have to schedule
                        // deallocation. Deferred drop is okay, because `list.delete()` can only be
                        // called if `T: 'static`.
                        unsafe {
                            C::finalize(self.curr.deref(), self.guard);
                        }

                        // `succ` is the new value of `self.pred`.
                        succ
                    }
                    Err(e) => {
                        // `e.current` is the current value of `self.pred`.
                        e.current
                    }
                };

                // If the predecessor node is already marked as deleted, we need to restart from
                // `head`.
                if succ.tag() != 0 {
                    self.pred = self.head;
                    self.curr = self.head.load(Acquire, self.guard);

                    return Some(Err(IterError::Stalled));
                }

                // Move over the removed by only advancing `curr`, not `pred`.
                self.curr = succ;
                continue;
            }

            // Move one step forward.
            self.pred = &c.next;
            self.curr = succ;

            return Some(Ok(unsafe { C::element_of(c) }));
        }

        // We reached the end of the list.
        None
    }
}

值得注意的是，逻辑上被删除的节点需要在遍历的过程中进行真正删除，删除逻辑如下图所示：