library/std/src/sync/mpsc/mpsc_queue.rs

   1 //! A mostly lock-free multi-producer, single consumer queue.
   2 //!
   3 //! This module contains an implementation of a concurrent MPSC queue. This
   4 //! queue can be used to share data between threads, and is also used as the
   5 //! building block of channels in rust.
   6 //!
   7 //! Note that the current implementation of this queue has a caveat of the `pop`
   8 //! method, and see the method for more information about it. Due to this
   9 //! caveat, this queue might not be appropriate for all use-cases.
  10
  11 // https://www.1024cores.net/home/lock-free-algorithms
  12 //                          /queues/non-intrusive-mpsc-node-based-queue
  13
  14 #[cfg(all(test, not(target_os = "emscripten")))]
  15 mod tests;
  16
  17 pub use self::PopResult::*;
  18
  19 use core::cell::UnsafeCell;
  20 use core::ptr;
  21
  22 use crate::boxed::Box;
  23 use crate::sync::atomic::{AtomicPtr, Ordering};
  24
  25 /// A result of the `pop` function.
  26 pub enum PopResult<T> {
  27     /// Some data has been popped
  28     Data(T),
  29     /// The queue is empty
  30     Empty,
  31     /// The queue is in an inconsistent state. Popping data should succeed, but
  32     /// some pushers have yet to make enough progress in order allow a pop to
  33     /// succeed. It is recommended that a pop() occur "in the near future" in
  34     /// order to see if the sender has made progress or not
  35     Inconsistent,
  36 }
  37
  38 struct Node<T> {
  39     next: AtomicPtr<Node<T>>,
  40     value: Option<T>,
  41 }
  42
  43 /// The multi-producer single-consumer structure. This is not cloneable, but it
  44 /// may be safely shared so long as it is guaranteed that there is only one
  45 /// popper at a time (many pushers are allowed).
  46 pub struct Queue<T> {
  47     head: AtomicPtr<Node<T>>,
  48     tail: UnsafeCell<*mut Node<T>>,
  49 }
  50
  51 unsafe impl<T: Send> Send for Queue<T> {}
  52 unsafe impl<T: Send> Sync for Queue<T> {}
  53
  54 impl<T> Node<T> {
  55     unsafe fn new(v: Option<T>) -> *mut Node<T> {
  56         Box::into_raw(box Node { next: AtomicPtr::new(ptr::null_mut()), value: v })
  57     }
  58 }
  59
  60 impl<T> Queue<T> {
  61     /// Creates a new queue that is safe to share among multiple producers and
  62     /// one consumer.
  63     pub fn new() -> Queue<T> {
  64         let stub = unsafe { Node::new(None) };
  65         Queue { head: AtomicPtr::new(stub), tail: UnsafeCell::new(stub) }
  66     }
  67
  68     /// Pushes a new value onto this queue.
  69     pub fn push(&self, t: T) {
  70         unsafe {
  71             let n = Node::new(Some(t));
  72             let prev = self.head.swap(n, Ordering::AcqRel);
  73             (*prev).next.store(n, Ordering::Release);
  74         }
  75     }
  76
  77     /// Pops some data from this queue.
  78     ///
  79     /// Note that the current implementation means that this function cannot
  80     /// return `Option<T>`. It is possible for this queue to be in an
  81     /// inconsistent state where many pushes have succeeded and completely
  82     /// finished, but pops cannot return `Some(t)`. This inconsistent state
  83     /// happens when a pusher is pre-empted at an inopportune moment.
  84     ///
  85     /// This inconsistent state means that this queue does indeed have data, but
  86     /// it does not currently have access to it at this time.
  87     pub fn pop(&self) -> PopResult<T> {
  88         unsafe {
  89             let tail = *self.tail.get();
  90             let next = (*tail).next.load(Ordering::Acquire);
  91
  92             if !next.is_null() {
  93                 *self.tail.get() = next;
  94                 assert!((*tail).value.is_none());
  95                 assert!((*next).value.is_some());
  96                 let ret = (*next).value.take().unwrap();
  97                 let _: Box<Node<T>> = Box::from_raw(tail);
  98                 return Data(ret);
  99             }
 100
 101             if self.head.load(Ordering::Acquire) == tail { Empty } else { Inconsistent }
 102         }
 103     }
 104 }
 105
 106 impl<T> Drop for Queue<T> {
 107     fn drop(&mut self) {
 108         unsafe {
 109             let mut cur = *self.tail.get();
 110             while !cur.is_null() {
 111                 let next = (*cur).next.load(Ordering::Relaxed);
 112                 let _: Box<Node<T>> = Box::from_raw(cur);
 113                 cur = next;
 114             }
 115         }
 116     }
 117 }