Параллельная реализация связанной очереди без блокировок

Aug 19 2020

Недавно я узнал о параллелизме / параллелизме и решил реализовать на практике безблокировочную связанную очередь Майкла и Скотта (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);
    }
}

Ответы

3 mpoeter Aug 21 2020 at 08:40

Я плохо владею 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.

Возможно, вы захотите взглянуть на мою реализацию очереди Майкла Скотта: https://github.com/mpoeter/xenium/blob/master/xenium/michael_scott_queue.hpp
Это сделано на C ++, но для решения проблемы восстановления памяти используется аналогичная концепция защиты.