1 //! A mostly lock-free multi-producer, single consumer queue.
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.
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.
11 // https://www.1024cores.net/home/lock-free-algorithms
12 // /queues/non-intrusive-mpsc-node-based-queue
14 #[cfg(all(test, not(target_os = "emscripten")))]
17 pub use self::PopResult::*;
19 use core::cell::UnsafeCell;
22 use crate::boxed::Box;
23 use crate::sync::atomic::{AtomicPtr, Ordering};
25 /// A result of the `pop` function.
26 pub enum PopResult<T> {
27 /// Some data has been popped
29 /// The queue is 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
39 next: AtomicPtr<Node<T>>,
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).
47 head: AtomicPtr<Node<T>>,
48 tail: UnsafeCell<*mut Node<T>>,
51 unsafe impl<T: Send> Send for Queue<T> {}
52 unsafe impl<T: Send> Sync for Queue<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 })
61 /// Creates a new queue that is safe to share among multiple producers and
63 pub fn new() -> Queue<T> {
64 let stub = unsafe { Node::new(None) };
65 Queue { head: AtomicPtr::new(stub), tail: UnsafeCell::new(stub) }
68 /// Pushes a new value onto this queue.
69 pub fn push(&self, t: T) {
71 let n = Node::new(Some(t));
72 let prev = self.head.swap(n, Ordering::AcqRel);
73 (*prev).next.store(n, Ordering::Release);
77 /// Pops some data from this queue.
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.
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> {
89 let tail = *self.tail.get();
90 let next = (*tail).next.load(Ordering::Acquire);
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);
101 if self.head.load(Ordering::Acquire) == tail { Empty } else { Inconsistent }
106 impl<T> Drop for Queue<T> {
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);