src/libsync/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 tasks, 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 #![experimental]
  39
  40 // http://www.1024cores.net/home/lock-free-algorithms
  41 //                         /queues/non-intrusive-mpsc-node-based-queue
  42
  43 use core::prelude::*;
  44
  45 use alloc::owned::Box;
  46 use core::mem;
  47 use core::ty::Unsafe;
  48
  49 use atomics::{AtomicPtr, Release, Acquire, AcqRel, Relaxed};
  50
  51 /// A result of the `pop` function.
  52 pub 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<Node<T>>,
  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 struct Queue<T> {
  73     head: AtomicPtr<Node<T>>,
  74     tail: Unsafe<*mut Node<T>>,
  75 }
  76
  77 impl<T> Node<T> {
  78     unsafe fn new(v: Option<T>) -> *mut Node<T> {
  79         mem::transmute(box Node {
  80             next: AtomicPtr::new(0 as *mut Node<T>),
  81             value: v,
  82         })
  83     }
  84 }
  85
  86 impl<T: Send> Queue<T> {
  87     /// Creates a new queue that is safe to share among multiple producers and
  88     /// one consumer.
  89     pub fn new() -> Queue<T> {
  90         let stub = unsafe { Node::new(None) };
  91         Queue {
  92             head: AtomicPtr::new(stub),
  93             tail: Unsafe::new(stub),
  94         }
  95     }
  96
  97     /// Pushes a new value onto this queue.
  98     pub fn push(&self, t: T) {
  99         unsafe {
 100             let n = Node::new(Some(t));
 101             let prev = self.head.swap(n, AcqRel);
 102             (*prev).next.store(n, Release);
 103         }
 104     }
 105
 106     /// Pops some data from this queue.
 107     ///
 108     /// Note that the current implementation means that this function cannot
 109     /// return `Option<T>`. It is possible for this queue to be in an
 110     /// inconsistent state where many pushes have succeeded and completely
 111     /// finished, but pops cannot return `Some(t)`. This inconsistent state
 112     /// happens when a pusher is pre-empted at an inopportune moment.
 113     ///
 114     /// This inconsistent state means that this queue does indeed have data, but
 115     /// it does not currently have access to it at this time.
 116     pub fn pop(&self) -> PopResult<T> {
 117         unsafe {
 118             let tail = *self.tail.get();
 119             let next = (*tail).next.load(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                 let _: Box<Node<T>> = mem::transmute(tail);
 127                 return Data(ret);
 128             }
 129
 130             if self.head.load(Acquire) == tail {Empty} else {Inconsistent}
 131         }
 132     }
 133
 134     /// Attempts to pop data from this queue, but doesn't attempt too hard. This
 135     /// will canonicalize inconsistent states to a `None` value.
 136     pub fn casual_pop(&self) -> Option<T> {
 137         match self.pop() {
 138             Data(t) => Some(t),
 139             Empty | Inconsistent => None,
 140         }
 141     }
 142 }
 143
 144 #[unsafe_destructor]
 145 impl<T: Send> Drop for Queue<T> {
 146     fn drop(&mut self) {
 147         unsafe {
 148             let mut cur = *self.tail.get();
 149             while !cur.is_null() {
 150                 let next = (*cur).next.load(Relaxed);
 151                 let _: Box<Node<T>> = mem::transmute(cur);
 152                 cur = next;
 153             }
 154         }
 155     }
 156 }
 157
 158 #[cfg(test)]
 159 mod tests {
 160     use std::prelude::*;
 161
 162     use alloc::arc::Arc;
 163
 164     use native;
 165     use super::{Queue, Data, Empty, Inconsistent};
 166
 167     #[test]
 168     fn test_full() {
 169         let q = Queue::new();
 170         q.push(box 1i);
 171         q.push(box 2i);
 172     }
 173
 174     #[test]
 175     fn test() {
 176         let nthreads = 8u;
 177         let nmsgs = 1000u;
 178         let q = Queue::new();
 179         match q.pop() {
 180             Empty => {}
 181             Inconsistent | Data(..) => fail!()
 182         }
 183         let (tx, rx) = channel();
 184         let q = Arc::new(q);
 185
 186         for _ in range(0, nthreads) {
 187             let tx = tx.clone();
 188             let q = q.clone();
 189             native::task::spawn(proc() {
 190                 for i in range(0, nmsgs) {
 191                     q.push(i);
 192                 }
 193                 tx.send(());
 194             });
 195         }
 196
 197         let mut i = 0u;
 198         while i < nthreads * nmsgs {
 199             match q.pop() {
 200                 Empty | Inconsistent => {},
 201                 Data(_) => { i += 1 }
 202             }
 203         }
 204         drop(tx);
 205         for _ in range(0, nthreads) {
 206             rx.recv();
 207         }
 208     }
 209 }