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