From 5876e21225f0cf34e8caa40b18db56fa716e8c92 Mon Sep 17 00:00:00 2001 From: Jason Toffaletti Date: Mon, 7 Oct 2013 00:07:04 -0700 Subject: add multi-producer multi-consumer bounded queue to use for sleeper list --- src/libstd/rt/mpmc_bounded_queue.rs | 199 ++++++++++++++++++++++++++++++++++++ 1 file changed, 199 insertions(+) create mode 100644 src/libstd/rt/mpmc_bounded_queue.rs (limited to 'src/libstd/rt/mpmc_bounded_queue.rs') diff --git a/src/libstd/rt/mpmc_bounded_queue.rs b/src/libstd/rt/mpmc_bounded_queue.rs new file mode 100644 index 00000000000..8e6ac8f79c7 --- /dev/null +++ b/src/libstd/rt/mpmc_bounded_queue.rs @@ -0,0 +1,199 @@ +/* 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. + */ + +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 { + buffer: ~[Node], + mask: uint, + enqueue_pos: AtomicUint, + dequeue_pos: AtomicUint, +} + +struct Queue { + priv state: UnsafeArc>, +} + +impl State { + fn with_capacity(capacity: uint) -> State { + let capacity = if capacity < 2 || (capacity & (capacity - 1)) != 0 { + // use next power of 2 as capacity + 2f64.pow(&((capacity as f64).log2().floor()+1f64)) as uint + } else { + capacity + }; + let buffer = do vec::from_fn(capacity) |i:uint| { + Node{sequence:AtomicUint::new(i),value:None} + }; + State{ + buffer: buffer, + mask: capacity-1, + enqueue_pos: AtomicUint::new(0), + dequeue_pos: AtomicUint::new(0), + } + } + + 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()); + } + } +} -- cgit 1.4.1-3-g733a5 From c372fa55560f1cdfdcb566f3027689ba88c46da5 Mon Sep 17 00:00:00 2001 From: Jason Toffaletti Date: Mon, 7 Oct 2013 00:43:34 -0700 Subject: add padding to prevent false sharing --- src/libstd/rt/mpmc_bounded_queue.rs | 8 ++++++++ 1 file changed, 8 insertions(+) (limited to 'src/libstd/rt/mpmc_bounded_queue.rs') diff --git a/src/libstd/rt/mpmc_bounded_queue.rs b/src/libstd/rt/mpmc_bounded_queue.rs index 8e6ac8f79c7..c0c7d281136 100644 --- a/src/libstd/rt/mpmc_bounded_queue.rs +++ b/src/libstd/rt/mpmc_bounded_queue.rs @@ -40,10 +40,14 @@ struct Node { } 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 { @@ -62,10 +66,14 @@ impl State { 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], } } -- cgit 1.4.1-3-g733a5 From 5e91ac10b65a4c20915868d3af3c06f1d3d3cada Mon Sep 17 00:00:00 2001 From: Jason Toffaletti Date: Tue, 8 Oct 2013 08:09:13 -0700 Subject: minor --- src/libstd/rt/mpmc_bounded_queue.rs | 8 ++++++-- 1 file changed, 6 insertions(+), 2 deletions(-) (limited to 'src/libstd/rt/mpmc_bounded_queue.rs') diff --git a/src/libstd/rt/mpmc_bounded_queue.rs b/src/libstd/rt/mpmc_bounded_queue.rs index c0c7d281136..ea5d1050c18 100644 --- a/src/libstd/rt/mpmc_bounded_queue.rs +++ b/src/libstd/rt/mpmc_bounded_queue.rs @@ -57,8 +57,12 @@ struct Queue { impl State { fn with_capacity(capacity: uint) -> State { let capacity = if capacity < 2 || (capacity & (capacity - 1)) != 0 { - // use next power of 2 as capacity - 2f64.pow(&((capacity as f64).log2().floor()+1f64)) as uint + if capacity < 2 { + 2u + } else { + // use next power of 2 as capacity + 2f64.pow(&((capacity as f64).log2().ceil())) as uint + } } else { capacity }; -- cgit 1.4.1-3-g733a5 From 49d9135eeaefc5ab267c2ee2bf1c28e245320709 Mon Sep 17 00:00:00 2001 From: Brian Anderson Date: Fri, 25 Oct 2013 18:33:05 -0700 Subject: Tidy --- src/etc/licenseck.py | 2 ++ src/libstd/rt/mpmc_bounded_queue.rs | 6 +++--- src/libstd/rt/mpsc_queue.rs | 6 +++--- 3 files changed, 8 insertions(+), 6 deletions(-) (limited to 'src/libstd/rt/mpmc_bounded_queue.rs') diff --git a/src/etc/licenseck.py b/src/etc/licenseck.py index 1e0c541cd89..91231430d2a 100644 --- a/src/etc/licenseck.py +++ b/src/etc/licenseck.py @@ -76,6 +76,8 @@ exceptions = [ "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/mpmc_bounded_queue.rs b/src/libstd/rt/mpmc_bounded_queue.rs index ea5d1050c18..a8ef5364276 100644 --- a/src/libstd/rt/mpmc_bounded_queue.rs +++ b/src/libstd/rt/mpmc_bounded_queue.rs @@ -2,14 +2,14 @@ * 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 diff --git a/src/libstd/rt/mpsc_queue.rs b/src/libstd/rt/mpsc_queue.rs index 93e66838a42..38186aa7e8c 100644 --- a/src/libstd/rt/mpsc_queue.rs +++ b/src/libstd/rt/mpsc_queue.rs @@ -2,14 +2,14 @@ * 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 -- cgit 1.4.1-3-g733a5 From 1ce5081f4d7a8d636f67204e0e62fe0e9164b560 Mon Sep 17 00:00:00 2001 From: Brian Anderson Date: Fri, 25 Oct 2013 19:46:35 -0700 Subject: Add links to original mpmc and mpsc implementations --- src/libstd/rt/mpmc_bounded_queue.rs | 2 ++ src/libstd/rt/mpsc_queue.rs | 1 + 2 files changed, 3 insertions(+) (limited to 'src/libstd/rt/mpmc_bounded_queue.rs') diff --git a/src/libstd/rt/mpmc_bounded_queue.rs b/src/libstd/rt/mpmc_bounded_queue.rs index a8ef5364276..2f61a433983 100644 --- a/src/libstd/rt/mpmc_bounded_queue.rs +++ b/src/libstd/rt/mpmc_bounded_queue.rs @@ -26,6 +26,8 @@ * 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::*; diff --git a/src/libstd/rt/mpsc_queue.rs b/src/libstd/rt/mpsc_queue.rs index 38186aa7e8c..4ddd5e066cd 100644 --- a/src/libstd/rt/mpsc_queue.rs +++ b/src/libstd/rt/mpsc_queue.rs @@ -27,6 +27,7 @@ */ //! 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}; -- cgit 1.4.1-3-g733a5