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