Una implementación simultánea de Linked Queue sin bloqueos
Recientemente he estado aprendiendo sobre concurrencia / paralelismo y decidí implementar la cola enlazada sin bloqueo de Michael & Scott (PDF) como práctica.
No estoy completamente seguro de cómo probar esta estructura de datos o incluso si mi implementación es segura para la concurrencia, pero se agradece cualquier comentario.
#![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);
}
}
Respuestas
No hablo con fluidez Rust, por lo que no puedo comentar sobre la implementación general. Sin embargo, lo que puedo decir es que esta implementación no es segura para subprocesos, ya que contiene varias condiciones de carrera.
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);
Si dos subprocesos observan un puntero nulo tail
, ambos actualizan directamente head
/ tail
. Obviamente, esta es una condición de carrera. En su lugar, debe crear un nodo ficticio vacío durante la inicialización de la cola (es decir, la cola siempre debe contener al menos un nodo; está vacía si head == tail
).
No estoy seguro de a qué te refieres con este comentario:
// 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.
El guard
es parte del programa para el rescate (época de recuperación basado en este caso), y sólo se impide eliminar un nodo que todavía se pueda acceder por algún otro hilo. Pero no evita que la cola se cambie justo debajo de la nariz.
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);
¡No puede almacenar directamente el nuevo nodo en tail`s next! Esta también es una condición de carrera, ya que otros subprocesos podrían estar haciendo lo mismo, sobrescribiendo efectivamente los valores escritos por otros subprocesos. Tienes que usar un bucle CAS para eso.
Lo mismo ocurre con la actualización frontal dequeue
.
Es posible que desee echar un vistazo a mi implementación de la cola de Michael Scott: https://github.com/mpoeter/xenium/blob/master/xenium/michael_scott_queue.hpp
Está hecho en C ++, pero usa un concepto de protección similar para resolver el problema de recuperación de memoria.