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