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
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::boxed::Box;
47 use std::cell::UnsafeCell;
48 use std::ptr;
49 use std::sync::atomic::{AtomicPtr, Ordering};
50 use std::thread;
51 
52 /// A result of the `pop` function.
53 pub(super) enum PopResult<T> {
54     /// Some data has been popped
55     Data(T),
56     /// The queue is empty
57     Empty,
58     /// The queue is in an inconsistent state. Popping data should succeed, but
59     /// some pushers have yet to make enough progress in order allow a pop to
60     /// succeed. It is recommended that a pop() occur "in the near future" in
61     /// order to see if the sender has made progress or not
62     Inconsistent,
63 }
64 
65 struct Node<T> {
66     next: AtomicPtr<Self>,
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 pub(super) struct Queue<T> {
74     head: AtomicPtr<Node<T>>,
75     tail: UnsafeCell<*mut Node<T>>,
76 }
77 
78 unsafe impl<T: Send> Send for Queue<T> {}
79 unsafe impl<T: Send> Sync for Queue<T> {}
80 
81 impl<T> Node<T> {
new(v: Option<T>) -> *mut Self82     unsafe fn new(v: Option<T>) -> *mut Self {
83         Box::into_raw(Box::new(Self { next: AtomicPtr::new(ptr::null_mut()), value: v }))
84     }
85 }
86 
87 impl<T> Queue<T> {
88     /// Creates a new queue that is safe to share among multiple producers and
89     /// one consumer.
new() -> Self90     pub(super) fn new() -> Self {
91         let stub = unsafe { Node::new(None) };
92         Self { head: AtomicPtr::new(stub), tail: UnsafeCell::new(stub) }
93     }
94 
95     /// Pushes a new value onto this queue.
push(&self, t: T)96     pub(super) fn push(&self, t: T) {
97         unsafe {
98             let n = Node::new(Some(t));
99             let prev = self.head.swap(n, Ordering::AcqRel);
100             (*prev).next.store(n, Ordering::Release);
101         }
102     }
103 
104     /// Pops some data from this queue.
105     ///
106     /// Note that the current implementation means that this function cannot
107     /// return `Option<T>`. It is possible for this queue to be in an
108     /// inconsistent state where many pushes have succeeded and completely
109     /// finished, but pops cannot return `Some(t)`. This inconsistent state
110     /// happens when a pusher is preempted at an inopportune moment.
111     ///
112     /// This inconsistent state means that this queue does indeed have data, but
113     /// it does not currently have access to it at this time.
114     ///
115     /// This function is unsafe because only one thread can call it at a time.
pop(&self) -> PopResult<T>116     pub(super) unsafe fn pop(&self) -> PopResult<T> {
117         unsafe {
118             let tail = *self.tail.get();
119             let next = (*tail).next.load(Ordering::Acquire);
120 
121             if !next.is_null() {
122                 *self.tail.get() = next;
123                 assert!((*tail).value.is_none());
124                 assert!((*next).value.is_some());
125                 let ret = (*next).value.take().unwrap();
126                 drop(Box::from_raw(tail));
127                 return Data(ret);
128             }
129 
130             if self.head.load(Ordering::Acquire) == tail {
131                 Empty
132             } else {
133                 Inconsistent
134             }
135         }
136     }
137 
138     /// Pop an element similarly to `pop` function, but spin-wait on inconsistent
139     /// queue state instead of returning `Inconsistent`.
140     ///
141     /// This function is unsafe because only one thread can call it at a time.
pop_spin(&self) -> Option<T>142     pub(super) unsafe fn pop_spin(&self) -> Option<T> {
143         loop {
144             match unsafe { self.pop() } {
145                 Empty => return None,
146                 Data(t) => return Some(t),
147                 // Inconsistent means that there will be a message to pop
148                 // in a short time. This branch can only be reached if
149                 // values are being produced from another thread, so there
150                 // are a few ways that we can deal with this:
151                 //
152                 // 1) Spin
153                 // 2) thread::yield_now()
154                 // 3) task::current().unwrap() & return Pending
155                 //
156                 // For now, thread::yield_now() is used, but it would
157                 // probably be better to spin a few times then yield.
158                 Inconsistent => {
159                     thread::yield_now();
160                 }
161             }
162         }
163     }
164 }
165 
166 impl<T> Drop for Queue<T> {
drop(&mut self)167     fn drop(&mut self) {
168         unsafe {
169             let mut cur = *self.tail.get();
170             while !cur.is_null() {
171                 let next = (*cur).next.load(Ordering::Relaxed);
172                 drop(Box::from_raw(cur));
173                 cur = next;
174             }
175         }
176     }
177 }
178