동시 잠금없는 Linked Queue 구현

Question

최근에 동시성 / 병렬성에 대해 배우고 있으며 Michael & Scott 잠금없는 연결 대기열 (PDF) 을 연습 으로 구현하기로 결정했습니다 .

이 데이터 구조를 테스트하는 방법이나 내 구현이 동시성에 안전한지 확실하지 않지만 의견을 보내 주시면 감사하겠습니다.

#![crate_name = "cqi"]

//! # cqi
//!
//! `cqi` provides a concurrent, lock-free implementation of a Linked Queue. This implementation is modelled after the
//! classic algorithms described in Maged M. Michael's and Michael L. Scott's paper ["Simple, Fast, and Practical
//! Non-Blocking and Blocking Concurrent Queue Algorithms"](https://www.cs.rochester.edu/u/scott/papers/1996_PODC_queues.pdf).
//!
//! A Linked Queue is a FIFO (first-in-first-out) abstract data type that sequentially stores its elements. Like all
//! queues, `cqi`'s Linked Queue implementation allows for insertion and deletion in order `O(1)`, with the additional
//! benefit of atomic reads and writes across multiple threads.

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

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

impl<T> Node<T> {
    pub fn new(item: T) -> Self {
        Self {
            item,
            next: Atomic::null(),
        }
    }
}

pub struct LinkedQueue<T> {
    head: Atomic<Node<T>>,
    tail: Atomic<Node<T>>,
    collector: Collector,
}

impl<T> LinkedQueue<T> {
    pub fn new() -> Self {
        LinkedQueue {
            head: Atomic::null(),
            tail: Atomic::null(),
            collector: epoch::default_collector().clone(),
        }
    }

    /// Retrieves a thread guard for the current thread. While the given guard is still in scope, any operations that
    /// involve mutating the queue will collect "garbage". This "garbage" is not freed until the guard has been dropped.
    /// Either manually drop the `Guard` or let it fall out of scope to prevent a lot of garbage from piling up.
    ///
    /// # Example
    /// ```
    /// use cqi::LinkedQueue;
    ///
    /// let lq = LinkedQueue::<usize>::new();
    /// let guard = lq.guard();
    /// ```
    pub fn guard(&self) -> Guard {
        self.collector.register().pin()
    }

    /// Inserts a new item at the back of the queue.
    ///
    /// # Example
    /// ```
    /// use cqi::LinkedQueue;
    ///
    /// let lq = LinkedQueue::<usize>::new();
    /// let guard = lq.guard();
    /// lq.enqueue(42, &guard);
    /// lq.enqueue(69, &guard);
    /// assert_eq!(lq.peek(&guard), Some(&42));
    /// ```
    pub fn enqueue<'g>(&self, item: T, guard: &'g Guard) {
        let new_node = Owned::new(Node::new(item)).into_shared(guard);

        // Unlike the enqueue algorithm described in M&S's paper, we don't need to check if the tail is consistent
        // between now and our CAS on the tail. Our `guard` ensures this.
        let tail = self.tail.load(Ordering::Acquire, guard);
        if tail.is_null() {
            self.head.store(new_node, Ordering::Release);
            self.tail.store(new_node, Ordering::Release);
        } else {
            let mut tail_node = unsafe { tail.deref() };
            let mut next = tail_node.next.load(Ordering::Acquire, guard);

            // Here we swing the tail forward if the last node in the queue is not the current node.
            while !next.is_null() {
                tail_node = unsafe { next.deref() };
                next = tail_node.next.load(Ordering::Acquire, guard);
            }

            tail_node.next.store(new_node, Ordering::Release);
            let _ = self
                .tail
                .compare_and_set(tail, new_node, Ordering::Release, guard);
        }
    }

    /// Removes the first item of the queue.
    ///
    /// # Example
    /// ```
    /// use cqi::LinkedQueue;
    ///
    /// let lq = LinkedQueue::<usize>::new();
    /// let guard = lq.guard();
    /// lq.enqueue(42, &guard);
    /// assert_eq!(lq.peek(&guard), Some(&42));
    /// lq.dequeue(&guard);
    /// assert_eq!(lq.peek(&guard), None);
    /// ```
    pub fn dequeue<'g>(&self, guard: &'g Guard) -> bool {
        let head = self.head.load(Ordering::Acquire, guard);

        if !head.is_null() {
            let head_node = unsafe { head.deref() };
            let next = head_node.next.load(Ordering::Acquire, guard);
            self.head.store(next, Ordering::Release);
            return true;
        }

        false
    }

    /// Retrieves the first item in the queue.
    ///
    /// # Example
    /// ```
    /// use cqi::LinkedQueue;
    ///
    /// let lq = LinkedQueue::<usize>::new();
    /// let guard = lq.guard();
    /// lq.enqueue(42, &guard);
    /// assert_eq!(lq.peek(&guard), Some(&42));
    /// ```
    pub fn peek<'g>(&self, guard: &'g Guard) -> Option<&'g T> {
        // Here we don't need to update the `mod_count` field in the `tail` node since we aren't doing any mutations.
        let head = self.head.load(Ordering::Acquire, guard);
        if head.is_null() {
            None
        } else {
            let item = unsafe { &head.deref().item };
            Some(item)
        }
    }

    /// Retrieves and removes the first item in the queue. **This operation can be expensive** as it copies the value
    /// being polled so it can be returned outside of the queue. Large types can impact performance here.
    ///
    /// # Example
    /// ```
    /// use cqi::LinkedQueue;
    ///
    /// let lq = LinkedQueue::<usize>::new();
    /// let guard = lq.guard();
    /// lq.enqueue(42, &guard);
    /// let item = lq.poll(&guard);
    ///
    /// assert_eq!(item, Some(42));
    /// assert_eq!(lq.peek(&guard), None);
    /// ```
    pub fn poll<'g>(&self, guard: &'g Guard) -> Option<T>
    where
        T: Copy,
    {
        let head = self.head.load(Ordering::Acquire, guard).to_owned();
        if head.is_null() {
            None
        } else {
            unsafe {
                let head_node = head.deref();
                let item = head_node.item.clone();
                self.head.store(
                    head_node.next.load(Ordering::Acquire, guard),
                    Ordering::Release,
                );
                Some(item)
            }
        }
    }

    /// Retrieves the number of items currently in the queue.
    ///
    /// As the queue can be concurrently updated, this will return the number of items in queue **at the time this
    /// function is called**. This number cannot be heavily relied on as it can already be out of date directly after
    /// this function is called.
    ///
    /// # Example
    /// ```
    /// use cqi::LinkedQueue;
    ///
    /// let lq = LinkedQueue::<usize>::new();
    /// let guard = lq.guard();
    /// lq.enqueue(42, &guard);
    /// lq.enqueue(69, &guard);
    /// assert_eq!(lq.len(&guard), 2);
    /// ```
    pub fn len<'g>(&self, guard: &'g Guard) -> usize {
        let mut size: usize = 0;
        let mut head = self.head.load(Ordering::SeqCst, guard);
        while !head.is_null() {
            size += 1;
            head = unsafe { head.deref().next.load(Ordering::SeqCst, guard) };
        }
        size
    }
}

#[cfg(test)]
mod tests {
    use super::LinkedQueue;

    #[test]
    fn test_enqueue() {
        let lq = LinkedQueue::<usize>::new();
        let guard = lq.guard();
        lq.enqueue(42, &guard);
        assert_eq!(lq.peek(&guard), Some(&42));

        lq.enqueue(69, &guard);
        assert_eq!(lq.peek(&guard), Some(&42));

        let _ = lq.poll(&guard);
        assert_eq!(lq.peek(&guard), Some(&69));
    }

    #[test]
    fn test_poll() {
        let lq = LinkedQueue::<usize>::new();
        let guard = lq.guard();
        lq.enqueue(42, &guard);
        lq.enqueue(69, &guard);

        // Ensure the item polled and the new head of the queue are the correct items.
        assert_eq!(lq.poll(&guard), Some(42));
        assert_eq!(lq.peek(&guard), Some(&69));
    }

    #[test]
    fn test_dequeue() {
        let lq = LinkedQueue::<usize>::new();
        let guard = lq.guard();
        lq.enqueue(42, &guard);
        lq.enqueue(69, &guard);

        lq.dequeue(&guard);
        assert_eq!(lq.peek(&guard), Some(&69));

        lq.dequeue(&guard);
        assert_eq!(lq.peek(&guard), None);
    }

    #[test]
    fn test_len() {
        let lq = LinkedQueue::<usize>::new();
        let guard = lq.guard();

        for i in 0..100 as usize {
            lq.enqueue(i, &guard);
        }

        assert_eq!(lq.len(&guard), 100);

        lq.dequeue(&guard);

        assert_eq!(lq.len(&guard), 99);

        for i in 0..99 as usize {
            lq.dequeue(&guard);
        }

        assert_eq!(lq.len(&guard), 0);
    }
}

Answer 1

저는 Rust에 능통하지 않기 때문에 전반적인 구현에 대해 언급 할 수 없습니다. 그러나 내가 말할 수있는 것은이 구현이 여러 경쟁 조건을 포함하기 때문에 스레드로부터 안전 하지 않다는 것입니다.

let tail = self.tail.load(Ordering::Acquire, guard);
        if tail.is_null() {
            self.head.store(new_node, Ordering::Release);
            self.tail.store(new_node, Ordering::Release);

두 스레드가에서 널 포인터를 관찰하면 tail둘 다 head/를 직접 업데이트 tail합니다. 이것은 분명히 경쟁 조건입니다. 대신 큐를 초기화하는 동안 빈 더미 노드를 만들어야합니다 (즉, 큐는 항상 하나 이상의 노드를 보유해야합니다.이면 비어 있음 head == tail).

이 의견이 무엇을 의미하는지 잘 모르겠습니다.

// Unlike the enqueue algorithm described in M&S's paper, we don't need to check if the tail is consistent
// between now and our CAS on the tail. Our `guard` ensures this.

은 guard매립 방식 (신기원이 경우 기반 매립)의 일부이며, 만 여전히 다른 스레드에서 액세스 할 수있는 노드를 삭제하지 못하도록합니다. 그러나 꼬리가 코 바로 아래에서 바뀌는 것을 막지는 못합니다.

            let mut tail_node = unsafe { tail.deref() };
            let mut next = tail_node.next.load(Ordering::Acquire, guard);

            // Here we swing the tail forward if the last node in the queue is not the current node.
            while !next.is_null() {
                tail_node = unsafe { next.deref() };
                next = tail_node.next.load(Ordering::Acquire, guard);
            }

            // this is a race condition!!
            tail_node.next.store(new_node, Ordering::Release);
            let _ = self
                .tail
                .compare_and_set(tail, new_node, Ordering::Release, guard);

새로운 노드를 꼬리에 직접 저장할 수 없습니다! 다른 스레드가 동일한 작업을 수행하여 다른 스레드가 작성한 값을 효과적으로 덮어 쓸 수 있기 때문에 이것은 경쟁 조건이기도합니다. 이를 위해 CAS 루프를 사용해야합니다.

에서 헤드를 업데이트하는 경우에도 마찬가지입니다 dequeue.

Michael Scott 대기열 구현을 살펴볼 수 있습니다. https://github.com/mpoeter/xenium/blob/master/xenium/michael_scott_queue.hpp
C ++로 수행되지만 유사한 가드 개념을 사용하여 메모리 회수 문제를 해결합니다.