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::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), 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<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 #[derive(Debug)] 74 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> { new(v: Option<T>) -> *mut Self83 unsafe fn new(v: Option<T>) -> *mut Self { 84 Box::into_raw(Box::new(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. new() -> Self94 pub(super) fn new() -> Self { 95 let stub = unsafe { Node::new(None) }; 96 Self { 97 head: AtomicPtr::new(stub), 98 tail: UnsafeCell::new(stub), 99 } 100 } 101 102 /// Pushes a new value onto this queue. push(&self, t: T)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. pop(&self) -> PopResult<T>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. pop_spin(&self) -> Option<T>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> { drop(&mut self)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