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 may not be appropriate for all use-cases.
11 // http://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
);