src/libstd/sync/mpsc/mpsc_queue.rs

   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.
  29 //!
  30 //! This module contains an implementation of a concurrent MPSC queue. This
  31 //! queue can be used to share data between threads, and is also used as the
  32 //! building block of channels in rust.
  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 pub use self::PopResult::*;
  42
  43 #[cfg(stage0)]
  44 use core::prelude::v1::*;
  45
  46 use alloc::boxed::Box;
  47 use core::ptr;
  48 use core::cell::UnsafeCell;
  49
  50 use sync::atomic::{AtomicPtr, Ordering};
  51
  52 /// A result of the `pop` function.
  53 pub 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<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 pub 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> {
  82     unsafe fn new(v: Option<T>) -> *mut Node<T> {
  83         Box::into_raw(box Node {
  84             next: AtomicPtr::new(ptr::null_mut()),
  85             value: v,
  86         })
  87     }
  88 }
  89
  90 impl<T> Queue<T> {
  91     /// Creates a new queue that is safe to share among multiple producers and
  92     /// one consumer.
  93     pub fn new() -> Queue<T> {
  94         let stub = unsafe { Node::new(None) };
  95         Queue {
  96             head: AtomicPtr::new(stub),
  97             tail: UnsafeCell::new(stub),
  98         }
  99     }
 100
 101     /// Pushes a new value onto this queue.
 102     pub fn push(&self, t: T) {
 103         unsafe {
 104             let n = Node::new(Some(t));
 105             let prev = self.head.swap(n, Ordering::AcqRel);
 106             (*prev).next.store(n, Ordering::Release);
 107         }
 108     }
 109
 110     /// Pops some data from this queue.
 111     ///
 112     /// Note that the current implementation means that this function cannot
 113     /// return `Option<T>`. It is possible for this queue to be in an
 114     /// inconsistent state where many pushes have succeeded and completely
 115     /// finished, but pops cannot return `Some(t)`. This inconsistent state
 116     /// happens when a pusher is pre-empted at an inopportune moment.
 117     ///
 118     /// This inconsistent state means that this queue does indeed have data, but
 119     /// it does not currently have access to it at this time.
 120     pub fn pop(&self) -> PopResult<T> {
 121         unsafe {
 122             let tail = *self.tail.get();
 123             let next = (*tail).next.load(Ordering::Acquire);
 124
 125             if !next.is_null() {
 126                 *self.tail.get() = next;
 127                 assert!((*tail).value.is_none());
 128                 assert!((*next).value.is_some());
 129                 let ret = (*next).value.take().unwrap();
 130                 let _: Box<Node<T>> = Box::from_raw(tail);
 131                 return Data(ret);
 132             }
 133
 134             if self.head.load(Ordering::Acquire) == tail {Empty} else {Inconsistent}
 135         }
 136     }
 137 }
 138
 139 #[stable(feature = "rust1", since = "1.0.0")]
 140 impl<T> Drop for Queue<T> {
 141     fn drop(&mut self) {
 142         unsafe {
 143             let mut cur = *self.tail.get();
 144             while !cur.is_null() {
 145                 let next = (*cur).next.load(Ordering::Relaxed);
 146                 let _: Box<Node<T>> = Box::from_raw(cur);
 147                 cur = next;
 148             }
 149         }
 150     }
 151 }
 152
 153 #[cfg(test)]
 154 mod tests {
 155     use prelude::v1::*;
 156
 157     use sync::mpsc::channel;
 158     use super::{Queue, Data, Empty, Inconsistent};
 159     use sync::Arc;
 160     use thread;
 161
 162     #[test]
 163     fn test_full() {
 164         let q: Queue<Box<_>> = Queue::new();
 165         q.push(box 1);
 166         q.push(box 2);
 167     }
 168
 169     #[test]
 170     fn test() {
 171         let nthreads = 8;
 172         let nmsgs = 1000;
 173         let q = Queue::new();
 174         match q.pop() {
 175             Empty => {}
 176             Inconsistent | Data(..) => panic!()
 177         }
 178         let (tx, rx) = channel();
 179         let q = Arc::new(q);
 180
 181         for _ in 0..nthreads {
 182             let tx = tx.clone();
 183             let q = q.clone();
 184             thread::spawn(move|| {
 185                 for i in 0..nmsgs {
 186                     q.push(i);
 187                 }
 188                 tx.send(()).unwrap();
 189             });
 190         }
 191
 192         let mut i = 0;
 193         while i < nthreads * nmsgs {
 194             match q.pop() {
 195                 Empty | Inconsistent => {},
 196                 Data(_) => { i += 1 }
 197             }
 198         }
 199         drop(tx);
 200         for _ in 0..nthreads {
 201             rx.recv().unwrap();
 202         }
 203     }
 204 }