Skip to main content
added 96 characters in body; edited title
Source Link
Zeta
  • 19.3k
  • 2
  • 55
  • 89

A concurrent lock-free Linked Queue implementation (Rust)

recentlyRecently I've been learning about concurrency/parallelism and I decided to implement the Michael&Scott lock-free Linked queueMichael&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 correctconcurrency 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);
    }
}

```

A concurrent lock-free Linked Queue implementation (Rust)

recently I've been learning about concurrency/parallelism and I decided to implement the Michael&Scott lock-free Linked queue as practice. I'm not entirely sure how to test this data structure or even if my implementation is correct, 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);
    }
}

```

A concurrent lock-free Linked Queue implementation

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

Source Link

A concurrent lock-free Linked Queue implementation (Rust)

recently I've been learning about concurrency/parallelism and I decided to implement the Michael&Scott lock-free Linked queue as practice. I'm not entirely sure how to test this data structure or even if my implementation is correct, 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);
    }
}

```