diff --git a/src/etc/licenseck.py b/src/etc/licenseck.py index 1e0c541cd8927..91231430d2a0b 100644 --- a/src/etc/licenseck.py +++ b/src/etc/licenseck.py @@ -76,6 +76,8 @@ "rt/isaac/randport.cpp", # public domain "rt/isaac/rand.h", # public domain "rt/isaac/standard.h", # public domain + "libstd/rt/mpsc_queue.rs", # BSD + "libstd/rt/mpmc_bounded_queue.rs", # BSD ] def check_license(name, contents): diff --git a/src/libstd/rt/message_queue.rs b/src/libstd/rt/message_queue.rs index 99b5156b31955..10e457368f007 100644 --- a/src/libstd/rt/message_queue.rs +++ b/src/libstd/rt/message_queue.rs @@ -11,83 +11,45 @@ //! A concurrent queue that supports multiple producers and a //! single consumer. -use container::Container; use kinds::Send; use vec::OwnedVector; -use cell::Cell; -use option::*; -use unstable::sync::{UnsafeArc, LittleLock}; +use option::Option; use clone::Clone; +use rt::mpsc_queue::Queue; pub struct MessageQueue { - priv state: UnsafeArc> -} - -struct State { - count: uint, - queue: ~[T], - lock: LittleLock + priv queue: Queue } impl MessageQueue { pub fn new() -> MessageQueue { MessageQueue { - state: UnsafeArc::new(State { - count: 0, - queue: ~[], - lock: LittleLock::new() - }) + queue: Queue::new() } } + #[inline] pub fn push(&mut self, value: T) { - unsafe { - let value = Cell::new(value); - let state = self.state.get(); - do (*state).lock.lock { - (*state).count += 1; - (*state).queue.push(value.take()); - } - } + self.queue.push(value) } + #[inline] pub fn pop(&mut self) -> Option { - unsafe { - let state = self.state.get(); - do (*state).lock.lock { - if !(*state).queue.is_empty() { - (*state).count += 1; - Some((*state).queue.shift()) - } else { - None - } - } - } + self.queue.pop() } /// A pop that may sometimes miss enqueued elements, but is much faster /// to give up without doing any synchronization + #[inline] pub fn casual_pop(&mut self) -> Option { - unsafe { - let state = self.state.get(); - // NB: Unsynchronized check - if (*state).count == 0 { return None; } - do (*state).lock.lock { - if !(*state).queue.is_empty() { - (*state).count += 1; - Some((*state).queue.shift()) - } else { - None - } - } - } + self.queue.pop() } } impl Clone for MessageQueue { fn clone(&self) -> MessageQueue { MessageQueue { - state: self.state.clone() + queue: self.queue.clone() } } } diff --git a/src/libstd/rt/mod.rs b/src/libstd/rt/mod.rs index 5113c28aa085e..d87580c83bfdb 100644 --- a/src/libstd/rt/mod.rs +++ b/src/libstd/rt/mod.rs @@ -136,6 +136,12 @@ mod work_queue; /// A parallel queue. mod message_queue; +/// A mostly lock-free multi-producer, single consumer queue. +mod mpsc_queue; + +/// A lock-free multi-producer, multi-consumer bounded queue. +mod mpmc_bounded_queue; + /// A parallel data structure for tracking sleeping schedulers. mod sleeper_list; diff --git a/src/libstd/rt/mpmc_bounded_queue.rs b/src/libstd/rt/mpmc_bounded_queue.rs new file mode 100644 index 0000000000000..2f61a4339833a --- /dev/null +++ b/src/libstd/rt/mpmc_bounded_queue.rs @@ -0,0 +1,213 @@ +/* Multi-producer/multi-consumer bounded queue + * Copyright (c) 2010-2011 Dmitry Vyukov. All rights reserved. + * Redistribution and use in source and binary forms, with or without + * modification, are permitted provided that the following conditions are met: + * + * 1. Redistributions of source code must retain the above copyright notice, + * this list of conditions and the following disclaimer. + * + * 2. Redistributions in binary form must reproduce the above copyright + * notice, this list of conditions and the following disclaimer in the + * documentation and/or other materials provided with the distribution. + * + * THIS SOFTWARE IS PROVIDED BY DMITRY VYUKOV "AS IS" AND ANY EXPRESS OR IMPLIED + * WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF + * MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT + * SHALL DMITRY VYUKOV OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, + * INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT + * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR + * PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF + * LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE + * OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF + * ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. + * + * The views and conclusions contained in the software and documentation are + * those of the authors and should not be interpreted as representing official + * policies, either expressed or implied, of Dmitry Vyukov. + */ + +// http://www.1024cores.net/home/lock-free-algorithms/queues/bounded-mpmc-queue + +use unstable::sync::UnsafeArc; +use unstable::atomics::{AtomicUint,Relaxed,Release,Acquire}; +use option::*; +use vec; +use clone::Clone; +use kinds::Send; +use num::{Exponential,Algebraic,Round}; + +struct Node { + sequence: AtomicUint, + value: Option, +} + +struct State { + pad0: [u8, ..64], + buffer: ~[Node], + mask: uint, + pad1: [u8, ..64], + enqueue_pos: AtomicUint, + pad2: [u8, ..64], + dequeue_pos: AtomicUint, + pad3: [u8, ..64], +} + +struct Queue { + priv state: UnsafeArc>, +} + +impl State { + fn with_capacity(capacity: uint) -> State { + let capacity = if capacity < 2 || (capacity & (capacity - 1)) != 0 { + if capacity < 2 { + 2u + } else { + // use next power of 2 as capacity + 2f64.pow(&((capacity as f64).log2().ceil())) as uint + } + } else { + capacity + }; + let buffer = do vec::from_fn(capacity) |i:uint| { + Node{sequence:AtomicUint::new(i),value:None} + }; + State{ + pad0: [0, ..64], + buffer: buffer, + mask: capacity-1, + pad1: [0, ..64], + enqueue_pos: AtomicUint::new(0), + pad2: [0, ..64], + dequeue_pos: AtomicUint::new(0), + pad3: [0, ..64], + } + } + + fn push(&mut self, value: T) -> bool { + let mask = self.mask; + let mut pos = self.enqueue_pos.load(Relaxed); + loop { + let node = &mut self.buffer[pos & mask]; + let seq = node.sequence.load(Acquire); + let diff: int = seq as int - pos as int; + + if diff == 0 { + let enqueue_pos = self.enqueue_pos.compare_and_swap(pos, pos+1, Relaxed); + if enqueue_pos == pos { + node.value = Some(value); + node.sequence.store(pos+1, Release); + break + } else { + pos = enqueue_pos; + } + } else if (diff < 0) { + return false + } else { + pos = self.enqueue_pos.load(Relaxed); + } + } + true + } + + fn pop(&mut self) -> Option { + let mask = self.mask; + let mut pos = self.dequeue_pos.load(Relaxed); + loop { + let node = &mut self.buffer[pos & mask]; + let seq = node.sequence.load(Acquire); + let diff: int = seq as int - (pos + 1) as int; + if diff == 0 { + let dequeue_pos = self.dequeue_pos.compare_and_swap(pos, pos+1, Relaxed); + if dequeue_pos == pos { + let value = node.value.take(); + node.sequence.store(pos + mask + 1, Release); + return value + } else { + pos = dequeue_pos; + } + } else if diff < 0 { + return None + } else { + pos = self.dequeue_pos.load(Relaxed); + } + } + } +} + +impl Queue { + pub fn with_capacity(capacity: uint) -> Queue { + Queue{ + state: UnsafeArc::new(State::with_capacity(capacity)) + } + } + + pub fn push(&mut self, value: T) -> bool { + unsafe { (*self.state.get()).push(value) } + } + + pub fn pop(&mut self) -> Option { + unsafe { (*self.state.get()).pop() } + } +} + +impl Clone for Queue { + fn clone(&self) -> Queue { + Queue { + state: self.state.clone() + } + } +} + +#[cfg(test)] +mod tests { + use prelude::*; + use option::*; + use task; + use comm; + use super::Queue; + + #[test] + fn test() { + let nthreads = 8u; + let nmsgs = 1000u; + let mut q = Queue::with_capacity(nthreads*nmsgs); + assert_eq!(None, q.pop()); + + for _ in range(0, nthreads) { + let (port, chan) = comm::stream(); + chan.send(q.clone()); + do task::spawn_sched(task::SingleThreaded) { + let mut q = port.recv(); + for i in range(0, nmsgs) { + assert!(q.push(i)); + } + } + } + + let mut completion_ports = ~[]; + for _ in range(0, nthreads) { + let (completion_port, completion_chan) = comm::stream(); + completion_ports.push(completion_port); + let (port, chan) = comm::stream(); + chan.send(q.clone()); + do task::spawn_sched(task::SingleThreaded) { + let mut q = port.recv(); + let mut i = 0u; + loop { + match q.pop() { + None => {}, + Some(_) => { + i += 1; + if i == nmsgs { break } + } + } + } + completion_chan.send(i); + } + } + + for completion_port in completion_ports.iter() { + assert_eq!(nmsgs, completion_port.recv()); + } + } +} diff --git a/src/libstd/rt/mpsc_queue.rs b/src/libstd/rt/mpsc_queue.rs new file mode 100644 index 0000000000000..4f39a1df4fa53 --- /dev/null +++ b/src/libstd/rt/mpsc_queue.rs @@ -0,0 +1,205 @@ +/* Multi-producer/single-consumer queue + * Copyright (c) 2010-2011 Dmitry Vyukov. All rights reserved. + * Redistribution and use in source and binary forms, with or without + * modification, are permitted provided that the following conditions are met: + * + * 1. Redistributions of source code must retain the above copyright notice, + * this list of conditions and the following disclaimer. + * + * 2. Redistributions in binary form must reproduce the above copyright + * notice, this list of conditions and the following disclaimer in the + * documentation and/or other materials provided with the distribution. + * + * THIS SOFTWARE IS PROVIDED BY DMITRY VYUKOV "AS IS" AND ANY EXPRESS OR IMPLIED + * WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF + * MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT + * SHALL DMITRY VYUKOV OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, + * INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT + * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR + * PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF + * LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE + * OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF + * ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. + * + * The views and conclusions contained in the software and documentation are + * those of the authors and should not be interpreted as representing official + * policies, either expressed or implied, of Dmitry Vyukov. + */ + +//! A mostly lock-free multi-producer, single consumer queue. +// http://www.1024cores.net/home/lock-free-algorithms/queues/intrusive-mpsc-node-based-queue + +use unstable::sync::UnsafeArc; +use unstable::atomics::{AtomicPtr,Relaxed,Release,Acquire}; +use ptr::{mut_null, to_mut_unsafe_ptr}; +use cast; +use option::*; +use clone::Clone; +use kinds::Send; + +struct Node { + next: AtomicPtr>, + value: Option, +} + +impl Node { + fn empty() -> Node { + Node{next: AtomicPtr::new(mut_null()), value: None} + } + + fn with_value(value: T) -> Node { + Node{next: AtomicPtr::new(mut_null()), value: Some(value)} + } +} + +struct State { + pad0: [u8, ..64], + head: AtomicPtr>, + pad1: [u8, ..64], + stub: Node, + pad2: [u8, ..64], + tail: *mut Node, + pad3: [u8, ..64], +} + +struct Queue { + priv state: UnsafeArc>, +} + +impl Clone for Queue { + fn clone(&self) -> Queue { + Queue { + state: self.state.clone() + } + } +} + +impl State { + pub fn new() -> State { + State{ + pad0: [0, ..64], + head: AtomicPtr::new(mut_null()), + pad1: [0, ..64], + stub: Node::::empty(), + pad2: [0, ..64], + tail: mut_null(), + pad3: [0, ..64], + } + } + + fn init(&mut self) { + let stub = self.get_stub_unsafe(); + self.head.store(stub, Relaxed); + self.tail = stub; + } + + fn get_stub_unsafe(&mut self) -> *mut Node { + to_mut_unsafe_ptr(&mut self.stub) + } + + fn push(&mut self, value: T) { + unsafe { + let node = cast::transmute(~Node::with_value(value)); + self.push_node(node); + } + } + + fn push_node(&mut self, node: *mut Node) { + unsafe { + (*node).next.store(mut_null(), Release); + let prev = self.head.swap(node, Relaxed); + (*prev).next.store(node, Release); + } + } + + fn pop(&mut self) -> Option { + unsafe { + let mut tail = self.tail; + let mut next = (*tail).next.load(Acquire); + let stub = self.get_stub_unsafe(); + if tail == stub { + if mut_null() == next { + return None + } + self.tail = next; + tail = next; + next = (*next).next.load(Acquire); + } + if next != mut_null() { + let tail: ~Node = cast::transmute(tail); + self.tail = next; + return tail.value + } + let head = self.head.load(Relaxed); + if tail != head { + return None + } + self.push_node(stub); + next = (*tail).next.load(Acquire); + if next != mut_null() { + let tail: ~Node = cast::transmute(tail); + self.tail = next; + return tail.value + } + } + None + } +} + +impl Queue { + pub fn new() -> Queue { + unsafe { + let q = Queue{state: UnsafeArc::new(State::new())}; + (*q.state.get()).init(); + q + } + } + + pub fn push(&mut self, value: T) { + unsafe { (*self.state.get()).push(value) } + } + + pub fn pop(&mut self) -> Option { + unsafe{ (*self.state.get()).pop() } + } +} + +#[cfg(test)] +mod tests { + use prelude::*; + use option::*; + use task; + use comm; + use super::Queue; + + #[test] + fn test() { + let nthreads = 8u; + let nmsgs = 1000u; + let mut q = Queue::new(); + assert_eq!(None, q.pop()); + + for _ in range(0, nthreads) { + let (port, chan) = comm::stream(); + chan.send(q.clone()); + do task::spawn_sched(task::SingleThreaded) { + let mut q = port.recv(); + for i in range(0, nmsgs) { + q.push(i); + } + } + } + + let mut i = 0u; + loop { + match q.pop() { + None => {}, + Some(_) => { + i += 1; + if i == nthreads*nmsgs { break } + } + } + } + } +} + diff --git a/src/libstd/rt/sleeper_list.rs b/src/libstd/rt/sleeper_list.rs index f4fdf15cda62e..39c7431837f6e 100644 --- a/src/libstd/rt/sleeper_list.rs +++ b/src/libstd/rt/sleeper_list.rs @@ -11,84 +11,37 @@ //! Maintains a shared list of sleeping schedulers. Schedulers //! use this to wake each other up. -use container::Container; -use vec::OwnedVector; -use option::{Option, Some, None}; -use cell::Cell; -use unstable::sync::{UnsafeArc, LittleLock}; use rt::sched::SchedHandle; +use rt::mpmc_bounded_queue::Queue; +use option::*; use clone::Clone; pub struct SleeperList { - priv state: UnsafeArc -} - -struct State { - count: uint, - stack: ~[SchedHandle], - lock: LittleLock + priv q: Queue, } impl SleeperList { pub fn new() -> SleeperList { - SleeperList { - state: UnsafeArc::new(State { - count: 0, - stack: ~[], - lock: LittleLock::new() - }) - } + SleeperList{q: Queue::with_capacity(8*1024)} } - pub fn push(&mut self, handle: SchedHandle) { - let handle = Cell::new(handle); - unsafe { - let state = self.state.get(); - do (*state).lock.lock { - (*state).count += 1; - (*state).stack.push(handle.take()); - } - } + pub fn push(&mut self, value: SchedHandle) { + assert!(self.q.push(value)) } pub fn pop(&mut self) -> Option { - unsafe { - let state = self.state.get(); - do (*state).lock.lock { - if !(*state).stack.is_empty() { - (*state).count -= 1; - Some((*state).stack.pop()) - } else { - None - } - } - } + self.q.pop() } - /// A pop that may sometimes miss enqueued elements, but is much faster - /// to give up without doing any synchronization pub fn casual_pop(&mut self) -> Option { - unsafe { - let state = self.state.get(); - // NB: Unsynchronized check - if (*state).count == 0 { return None; } - do (*state).lock.lock { - if !(*state).stack.is_empty() { - // NB: count is also protected by the lock - (*state).count -= 1; - Some((*state).stack.pop()) - } else { - None - } - } - } + self.q.pop() } } impl Clone for SleeperList { fn clone(&self) -> SleeperList { SleeperList { - state: self.state.clone() + q: self.q.clone() } } }