Un'implementazione simultanea della coda collegata priva di blocchi

Aug 19 2020

Recentemente ho imparato a conoscere la concorrenza / parallelismo e ho deciso di implementare la coda collegata senza blocchi di Michael & Scott (PDF) come pratica.

Non sono del tutto sicuro di come testare questa struttura di dati o anche se la mia implementazione è sicura per la concorrenza, ma qualsiasi feedback è apprezzato.

#![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);
    }
}

Risposte

3 mpoeter Aug 21 2020 at 08:40

Non parlo fluentemente Rust, quindi non posso commentare l'implementazione complessiva. Tuttavia, quello che posso dire è che questa implementazione non è thread-safe, poiché contiene diverse condizioni di competizione.

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);

Se due thread osservano un puntatore nullo in tail, entrambi aggiornano direttamente head/ tail. Questa è ovviamente una condizione di gara. Invece, è necessario creare un nodo fittizio vuoto durante l'inizializzazione della coda (cioè, la coda deve sempre contenere almeno un nodo; è vuota se head == tail).

Non sono sicuro di cosa intendi con questo commento:

// 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.

Fa guardparte dello schema di recupero (recupero basato sull'epoca in questo caso) e impedisce solo di eliminare un nodo a cui potrebbe ancora accedere un altro thread. Ma non impedisce che la coda venga cambiata proprio sotto il naso.

            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);

Non è possibile memorizzare direttamente il nuovo nodo in tail`s next! Questa è anche una condizione di competizione poiché altri thread potrebbero fare lo stesso, sovrascrivendo efficacemente i valori scritti da altri thread. Devi usare un ciclo CAS per questo.

Lo stesso vale per l'aggiornamento a testa alta dequeue.

Potresti dare un'occhiata alla mia implementazione della coda di Michael Scott: https://github.com/mpoeter/xenium/blob/master/xenium/michael_scott_queue.hpp
È fatto in C ++, ma utilizza un concetto di guardia simile per risolvere il problema di recupero della memoria.