4
\$\begingroup\$

Recently I've been learning about concurrency/parallelism and I decided to implement the Michael&Scott lock-free Linked queue (PDF) as practice.

I'm not entirely sure how to test this data structure or even if my implementation is concurrency safe, but any feedback is appreciated.

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

\$\endgroup\$
1
  • \$\begingroup\$ Welcome to CodeReview! I hope you get some nice reviews :) \$\endgroup\$
    – Zeta
    Aug 19 '20 at 18:26
3
\$\begingroup\$

I am not fluent in Rust, so I cannot comment on the overall implementation. However, what I can say is that this implementation is not thread-safe, as it contains several race conditions.

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

If two threads observe a null pointer in tail, both directly update head/tail. This is obviously a race condition. Instead, you need to create an empty dummy node during initialization of the queue (i.e., the queue always has to hold at least one node; it is empty if head == tail).

I am not sure what you mean by this comment:

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

The guard is part of the reclamation scheme (epoch based reclamation in this case), and it only prevents you from deleting a node that might still be accessed by some other thread. But it does not prevent tail from getting changed right under your nose.

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

You cannot directly store the new node into tail`s next! This is also a race condition since other threads might be doing the same, effectively overwritting the values written by some other threads. You have to use a CAS loop for that.

The same goes for updating head in dequeue.

You might want to take a look at my implementation of the Michael Scott queue: https://github.com/mpoeter/xenium/blob/master/xenium/michael_scott_queue.hpp
It is done in C++, but it uses a similar guard concept to solve the memory reclamation problem.

\$\endgroup\$
1
  • \$\begingroup\$ This really helps a lot. Thank you! \$\endgroup\$
    – chromaticc
    Aug 22 '20 at 9:57

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.