1 /* Copyright (c) 2010-2011 Dmitry Vyukov. All rights reserved.
2 * Redistribution and use in source and binary forms, with or without
3 * modification, are permitted provided that the following conditions are met:
4 *
5 * 1. Redistributions of source code must retain the above copyright notice,
6 * this list of conditions and the following disclaimer.
7 *
8 * 2. Redistributions in binary form must reproduce the above copyright
9 * notice, this list of conditions and the following disclaimer in the
10 * documentation and/or other materials provided with the distribution.
11 *
12 * THIS SOFTWARE IS PROVIDED BY DMITRY VYUKOV "AS IS" AND ANY EXPRESS OR IMPLIED
13 * WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF
14 * MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT
15 * SHALL DMITRY VYUKOV OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT,
16 * INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
17 * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR
18 * PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF
19 * LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE
20 * OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF
21 * ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
22 *
23 * The views and conclusions contained in the software and documentation are
24 * those of the authors and should not be interpreted as representing official
25 * policies, either expressed or implied, of Dmitry Vyukov.
26 */
27
28 //! A mostly lock-free multi-producer, single consumer queue for sending
29 //! messages between asynchronous tasks.
30 //!
31 //! The queue implementation is essentially the same one used for mpsc channels
32 //! in the standard library.
33 //!
34 //! Note that the current implementation of this queue has a caveat of the `pop`
35 //! method, and see the method for more information about it. Due to this
36 //! caveat, this queue may not be appropriate for all use-cases.
37
38 // http://www.1024cores.net/home/lock-free-algorithms
new(t: T) -> Lock<T>39 // /queues/non-intrusive-mpsc-node-based-queue
40
41 // NOTE: this implementation is lifted from the standard library and only
42 // slightly modified
43
44 pub(super) use self::PopResult::*;
45
46 use std::thread;
47 use std::cell::UnsafeCell;
48 use std::ptr;
49 use std::sync::atomic::{AtomicPtr, Ordering};
50
51 /// A result of the `pop` function.
52 pub(super) enum PopResult<T> {
53 /// Some data has been popped
54 Data(T),
try_lock(&self) -> Option<TryLock<'_, T>>55 /// The queue is empty
56 Empty,
57 /// The queue is in an inconsistent state. Popping data should succeed, but
58 /// some pushers have yet to make enough progress in order allow a pop to
59 /// succeed. It is recommended that a pop() occur "in the near future" in
60 /// order to see if the sender has made progress or not
61 Inconsistent,
62 }
63
64 #[derive(Debug)]
65 struct Node<T> {
66 next: AtomicPtr<Node<T>>,
67 value: Option<T>,
68 }
69
70 /// The multi-producer single-consumer structure. This is not cloneable, but it
71 /// may be safely shared so long as it is guaranteed that there is only one
72 /// popper at a time (many pushers are allowed).
73 #[derive(Debug)]
deref_mut(&mut self) -> &mut T74 pub(super) struct Queue<T> {
75 head: AtomicPtr<Node<T>>,
76 tail: UnsafeCell<*mut Node<T>>,
77 }
78
79 unsafe impl<T: Send> Send for Queue<T> { }
80 unsafe impl<T: Send> Sync for Queue<T> { }
81
82 impl<T> Node<T> {
83 unsafe fn new(v: Option<T>) -> *mut Node<T> {
84 Box::into_raw(Box::new(Node {
drop(&mut self)85 next: AtomicPtr::new(ptr::null_mut()),
86 value: v,
87 }))
88 }
89 }
90
91 impl<T> Queue<T> {
92 /// Creates a new queue that is safe to share among multiple producers and
93 /// one consumer.
94 pub(super) fn new() -> Queue<T> {
95 let stub = unsafe { Node::new(None) };
96 Queue {
97 head: AtomicPtr::new(stub),
98 tail: UnsafeCell::new(stub),
99 }
100 }
101
102 /// Pushes a new value onto this queue.
103 pub(super) fn push(&self, t: T) {
104 unsafe {
105 let n = Node::new(Some(t));
106 let prev = self.head.swap(n, Ordering::AcqRel);
107 (*prev).next.store(n, Ordering::Release);
108 }
109 }
110
111 /// Pops some data from this queue.
112 ///
113 /// Note that the current implementation means that this function cannot
114 /// return `Option<T>`. It is possible for this queue to be in an
115 /// inconsistent state where many pushes have succeeded and completely
116 /// finished, but pops cannot return `Some(t)`. This inconsistent state
117 /// happens when a pusher is preempted at an inopportune moment.
118 ///
119 /// This inconsistent state means that this queue does indeed have data, but
120 /// it does not currently have access to it at this time.
121 ///
122 /// This function is unsafe because only one thread can call it at a time.
123 pub(super) unsafe fn pop(&self) -> PopResult<T> {
124 let tail = *self.tail.get();
125 let next = (*tail).next.load(Ordering::Acquire);
126
127 if !next.is_null() {
128 *self.tail.get() = next;
129 assert!((*tail).value.is_none());
130 assert!((*next).value.is_some());
131 let ret = (*next).value.take().unwrap();
132 drop(Box::from_raw(tail));
133 return Data(ret);
134 }
135
136 if self.head.load(Ordering::Acquire) == tail {Empty} else {Inconsistent}
137 }
138
139 /// Pop an element similarly to `pop` function, but spin-wait on inconsistent
140 /// queue state instead of returning `Inconsistent`.
141 ///
142 /// This function is unsafe because only one thread can call it at a time.
143 pub(super) unsafe fn pop_spin(&self) -> Option<T> {
144 loop {
145 match self.pop() {
146 Empty => return None,
147 Data(t) => return Some(t),
148 // Inconsistent means that there will be a message to pop
149 // in a short time. This branch can only be reached if
150 // values are being produced from another thread, so there
151 // are a few ways that we can deal with this:
152 //
153 // 1) Spin
154 // 2) thread::yield_now()
155 // 3) task::current().unwrap() & return Pending
156 //
157 // For now, thread::yield_now() is used, but it would
158 // probably be better to spin a few times then yield.
159 Inconsistent => {
160 thread::yield_now();
161 }
162 }
163 }
164 }
165 }
166
167 impl<T> Drop for Queue<T> {
168 fn drop(&mut self) {
169 unsafe {
170 let mut cur = *self.tail.get();
171 while !cur.is_null() {
172 let next = (*cur).next.load(Ordering::Relaxed);
173 drop(Box::from_raw(cur));
174 cur = next;
175 }
176 }
177 }
178 }
179