From bfa9064ba2687eb1d95708f72f41ddd9729a6ba1 Mon Sep 17 00:00:00 2001 From: Alex Crichton Date: Thu, 5 Dec 2013 17:56:17 -0800 Subject: Rewrite std::comm * Streams are now ~3x faster than before (fewer allocations and more optimized) * Based on a single-producer single-consumer lock-free queue that doesn't always have to allocate on every send. * Blocking via mutexes/cond vars outside the runtime * Streams work in/out of the runtime seamlessly * Select now works in/out of the runtime seamlessly * Streams will now fail!() on send() if the other end has hung up * try_send() will not fail * PortOne/ChanOne removed * SharedPort removed * MegaPipe removed * Generic select removed (only one kind of port now) * API redesign * try_recv == never block * recv_opt == block, don't fail * iter() == Iterator for Port * removed peek * Type::new * Removed rt::comm --- src/libstd/rt/mpmc_bounded_queue.rs | 18 +-- src/libstd/rt/mpsc_queue.rs | 230 ++++++++++++++-------------- src/libstd/rt/spsc_queue.rs | 296 ++++++++++++++++++++++++++++++++++++ src/libstd/rt/task.rs | 1 - 4 files changed, 423 insertions(+), 122 deletions(-) create mode 100644 src/libstd/rt/spsc_queue.rs (limited to 'src/libstd/rt') diff --git a/src/libstd/rt/mpmc_bounded_queue.rs b/src/libstd/rt/mpmc_bounded_queue.rs index 7f607fcf12a..1e04e5eb78d 100644 --- a/src/libstd/rt/mpmc_bounded_queue.rs +++ b/src/libstd/rt/mpmc_bounded_queue.rs @@ -1,5 +1,4 @@ -/* Multi-producer/multi-consumer bounded queue - * Copyright (c) 2010-2011 Dmitry Vyukov. All rights reserved. +/* 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: * @@ -163,7 +162,6 @@ mod tests { use prelude::*; use option::*; use task; - use comm; use super::Queue; #[test] @@ -174,10 +172,9 @@ mod tests { assert_eq!(None, q.pop()); for _ in range(0, nthreads) { - let (port, chan) = comm::stream(); - chan.send(q.clone()); + let q = q.clone(); do task::spawn_sched(task::SingleThreaded) { - let mut q = port.recv(); + let mut q = q; for i in range(0, nmsgs) { assert!(q.push(i)); } @@ -186,12 +183,11 @@ mod tests { let mut completion_ports = ~[]; for _ in range(0, nthreads) { - let (completion_port, completion_chan) = comm::stream(); + let (completion_port, completion_chan) = Chan::new(); completion_ports.push(completion_port); - let (port, chan) = comm::stream(); - chan.send(q.clone()); + let q = q.clone(); do task::spawn_sched(task::SingleThreaded) { - let mut q = port.recv(); + let mut q = q; let mut i = 0u; loop { match q.pop() { @@ -206,7 +202,7 @@ mod tests { } } - for completion_port in completion_ports.iter() { + for completion_port in completion_ports.mut_iter() { assert_eq!(nmsgs, completion_port.recv()); } } diff --git a/src/libstd/rt/mpsc_queue.rs b/src/libstd/rt/mpsc_queue.rs index 4f39a1df4fa..d575028af70 100644 --- a/src/libstd/rt/mpsc_queue.rs +++ b/src/libstd/rt/mpsc_queue.rs @@ -1,5 +1,4 @@ -/* Multi-producer/single-consumer queue - * Copyright (c) 2010-2011 Dmitry Vyukov. All rights reserved. +/* 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: * @@ -27,163 +26,177 @@ */ //! 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}; +// http://www.1024cores.net/home/lock-free-algorithms +// /queues/non-intrusive-mpsc-node-based-queue + use cast; -use option::*; use clone::Clone; use kinds::Send; +use ops::Drop; +use option::{Option, None, Some}; +use unstable::atomics::{AtomicPtr, Release, Acquire, AcqRel, Relaxed}; +use unstable::sync::UnsafeArc; + +pub enum PopResult { + /// Some data has been popped + Data(T), + /// The queue is empty + Empty, + /// The queue is in an inconsistent state. Popping data should succeed, but + /// some pushers have yet to make enough progress in order allow a pop to + /// succeed. It is recommended that a pop() occur "in the near future" in + /// order to see if the sender has made progress or not + Inconsistent, +} 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], +struct State { head: AtomicPtr>, - pad1: [u8, ..64], - stub: Node, - pad2: [u8, ..64], tail: *mut Node, - pad3: [u8, ..64], + packet: P, } -struct Queue { - priv state: UnsafeArc>, +pub struct Consumer { + priv state: UnsafeArc>, } -impl Clone for Queue { - fn clone(&self) -> Queue { - Queue { - state: self.state.clone() - } - } +pub struct Producer { + priv state: UnsafeArc>, } -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], - } +impl Clone for Producer { + fn clone(&self) -> Producer { + Producer { state: self.state.clone() } } +} - fn init(&mut self) { - let stub = self.get_stub_unsafe(); - self.head.store(stub, Relaxed); - self.tail = stub; +pub fn queue(p: P) -> (Consumer, Producer) { + unsafe { + let (a, b) = UnsafeArc::new2(State::new(p)); + (Consumer { state: a }, Producer { state: b }) } +} - fn get_stub_unsafe(&mut self) -> *mut Node { - to_mut_unsafe_ptr(&mut self.stub) +impl Node { + unsafe fn new(v: Option) -> *mut Node { + cast::transmute(~Node { + next: AtomicPtr::new(0 as *mut Node), + value: v, + }) } +} - fn push(&mut self, value: T) { - unsafe { - let node = cast::transmute(~Node::with_value(value)); - self.push_node(node); +impl State { + pub unsafe fn new(p: P) -> State { + let stub = Node::new(None); + State { + head: AtomicPtr::new(stub), + tail: stub, + packet: p, } } - 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); - } + unsafe fn push(&mut self, t: T) { + let n = Node::new(Some(t)); + let prev = self.head.swap(n, AcqRel); + (*prev).next.store(n, 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 - } + unsafe fn pop(&mut self) -> PopResult { + let tail = self.tail; + let next = (*tail).next.load(Acquire); + + if !next.is_null() { + self.tail = next; + assert!((*tail).value.is_none()); + assert!((*next).value.is_some()); + let ret = (*next).value.take_unwrap(); + let _: ~Node = cast::transmute(tail); + return Data(ret); } - None + + if self.head.load(Acquire) == tail {Empty} else {Inconsistent} + } + + unsafe fn is_empty(&mut self) -> bool { + return (*self.tail).next.load(Acquire).is_null(); } } -impl Queue { - pub fn new() -> Queue { +#[unsafe_destructor] +impl Drop for State { + fn drop(&mut self) { unsafe { - let q = Queue{state: UnsafeArc::new(State::new())}; - (*q.state.get()).init(); - q + let mut cur = self.tail; + while !cur.is_null() { + let next = (*cur).next.load(Relaxed); + let _: ~Node = cast::transmute(cur); + cur = next; + } } } +} +impl Producer { pub fn push(&mut self, value: T) { unsafe { (*self.state.get()).push(value) } } + pub fn is_empty(&self) -> bool { + unsafe{ (*self.state.get()).is_empty() } + } + pub unsafe fn packet(&self) -> *mut P { + &mut (*self.state.get()).packet as *mut P + } +} - pub fn pop(&mut self) -> Option { - unsafe{ (*self.state.get()).pop() } +impl Consumer { + pub fn pop(&mut self) -> PopResult { + unsafe { (*self.state.get()).pop() } + } + pub fn casual_pop(&mut self) -> Option { + match self.pop() { + Data(t) => Some(t), + Empty | Inconsistent => None, + } + } + pub unsafe fn packet(&self) -> *mut P { + &mut (*self.state.get()).packet as *mut P } } #[cfg(test)] mod tests { use prelude::*; - use option::*; + use task; - use comm; - use super::Queue; + use super::{queue, Data, Empty, Inconsistent}; + + #[test] + fn test_full() { + let (_, mut p) = queue(()); + p.push(~1); + p.push(~2); + } #[test] fn test() { let nthreads = 8u; let nmsgs = 1000u; - let mut q = Queue::new(); - assert_eq!(None, q.pop()); + let (mut c, p) = queue(()); + match c.pop() { + Empty => {} + Inconsistent | Data(..) => fail!() + } for _ in range(0, nthreads) { - let (port, chan) = comm::stream(); - chan.send(q.clone()); + let q = p.clone(); do task::spawn_sched(task::SingleThreaded) { - let mut q = port.recv(); + let mut q = q; for i in range(0, nmsgs) { q.push(i); } @@ -191,13 +204,10 @@ mod tests { } let mut i = 0u; - loop { - match q.pop() { - None => {}, - Some(_) => { - i += 1; - if i == nthreads*nmsgs { break } - } + while i < nthreads * nmsgs { + match c.pop() { + Empty | Inconsistent => {}, + Data(_) => { i += 1 } } } } diff --git a/src/libstd/rt/spsc_queue.rs b/src/libstd/rt/spsc_queue.rs new file mode 100644 index 00000000000..f14533d726a --- /dev/null +++ b/src/libstd/rt/spsc_queue.rs @@ -0,0 +1,296 @@ +/* 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/unbounded-spsc-queue +use cast; +use kinds::Send; +use ops::Drop; +use option::{Some, None, Option}; +use unstable::atomics::{AtomicPtr, Relaxed, AtomicUint, Acquire, Release}; +use unstable::sync::UnsafeArc; + +// Node within the linked list queue of messages to send +struct Node { + // XXX: this could be an uninitialized T if we're careful enough, and + // that would reduce memory usage (and be a bit faster). + // is it worth it? + value: Option, // nullable for re-use of nodes + next: AtomicPtr>, // next node in the queue +} + +// The producer/consumer halves both need access to the `tail` field, and if +// they both have access to that we may as well just give them both access +// to this whole structure. +struct State { + // consumer fields + tail: *mut Node, // where to pop from + tail_prev: AtomicPtr>, // where to pop from + + // producer fields + head: *mut Node, // where to push to + first: *mut Node, // where to get new nodes from + tail_copy: *mut Node, // between first/tail + + // Cache maintenance fields. Additions and subtractions are stored + // separately in order to allow them to use nonatomic addition/subtraction. + cache_bound: uint, + cache_additions: AtomicUint, + cache_subtractions: AtomicUint, + + packet: P, +} + +pub struct Producer { + priv state: UnsafeArc>, +} + +pub struct Consumer { + priv state: UnsafeArc>, +} + +pub fn queue(bound: uint, + p: P) -> (Consumer, Producer) +{ + let n1 = Node::new(); + let n2 = Node::new(); + unsafe { (*n1).next.store(n2, Relaxed) } + let state = State { + tail: n2, + tail_prev: AtomicPtr::new(n1), + head: n2, + first: n1, + tail_copy: n1, + cache_bound: bound, + cache_additions: AtomicUint::new(0), + cache_subtractions: AtomicUint::new(0), + packet: p, + }; + let (arc1, arc2) = UnsafeArc::new2(state); + (Consumer { state: arc1 }, Producer { state: arc2 }) +} + +impl Node { + fn new() -> *mut Node { + unsafe { + cast::transmute(~Node { + value: None, + next: AtomicPtr::new(0 as *mut Node), + }) + } + } +} + +impl Producer { + pub fn push(&mut self, t: T) { + unsafe { (*self.state.get()).push(t) } + } + pub fn is_empty(&self) -> bool { + unsafe { (*self.state.get()).is_empty() } + } + pub unsafe fn packet(&self) -> *mut P { + &mut (*self.state.get()).packet as *mut P + } +} + +impl Consumer { + pub fn pop(&mut self) -> Option { + unsafe { (*self.state.get()).pop() } + } + pub unsafe fn packet(&self) -> *mut P { + &mut (*self.state.get()).packet as *mut P + } +} + +impl State { + // remember that there is only one thread executing `push` (and only one + // thread executing `pop`) + unsafe fn push(&mut self, t: T) { + // Acquire a node (which either uses a cached one or allocates a new + // one), and then append this to the 'head' node. + let n = self.alloc(); + assert!((*n).value.is_none()); + (*n).value = Some(t); + (*n).next.store(0 as *mut Node, Relaxed); + (*self.head).next.store(n, Release); + self.head = n; + } + + unsafe fn alloc(&mut self) -> *mut Node { + // First try to see if we can consume the 'first' node for our uses. + // We try to avoid as many atomic instructions as possible here, so + // the addition to cache_subtractions is not atomic (plus we're the + // only one subtracting from the cache). + if self.first != self.tail_copy { + if self.cache_bound > 0 { + let b = self.cache_subtractions.load(Relaxed); + self.cache_subtractions.store(b + 1, Relaxed); + } + let ret = self.first; + self.first = (*ret).next.load(Relaxed); + return ret; + } + // If the above fails, then update our copy of the tail and try + // again. + self.tail_copy = self.tail_prev.load(Acquire); + if self.first != self.tail_copy { + if self.cache_bound > 0 { + let b = self.cache_subtractions.load(Relaxed); + self.cache_subtractions.store(b + 1, Relaxed); + } + let ret = self.first; + self.first = (*ret).next.load(Relaxed); + return ret; + } + // If all of that fails, then we have to allocate a new node + // (there's nothing in the node cache). + Node::new() + } + + // remember that there is only one thread executing `pop` (and only one + // thread executing `push`) + unsafe fn pop(&mut self) -> Option { + // The `tail` node is not actually a used node, but rather a + // sentinel from where we should start popping from. Hence, look at + // tail's next field and see if we can use it. If we do a pop, then + // the current tail node is a candidate for going into the cache. + let tail = self.tail; + let next = (*tail).next.load(Acquire); + if next.is_null() { return None } + assert!((*next).value.is_some()); + let ret = (*next).value.take(); + + self.tail = next; + if self.cache_bound == 0 { + self.tail_prev.store(tail, Release); + } else { + // XXX: this is dubious with overflow. + let additions = self.cache_additions.load(Relaxed); + let subtractions = self.cache_subtractions.load(Relaxed); + let size = additions - subtractions; + + if size < self.cache_bound { + self.tail_prev.store(tail, Release); + self.cache_additions.store(additions + 1, Relaxed); + } else { + (*self.tail_prev.load(Relaxed)).next.store(next, Relaxed); + // We have successfully erased all references to 'tail', so + // now we can safely drop it. + let _: ~Node = cast::transmute(tail); + } + } + return ret; + } + + unsafe fn is_empty(&self) -> bool { + let tail = self.tail; + let next = (*tail).next.load(Acquire); + return next.is_null(); + } +} + +#[unsafe_destructor] +impl Drop for State { + fn drop(&mut self) { + unsafe { + let mut cur = self.first; + while !cur.is_null() { + let next = (*cur).next.load(Relaxed); + let _n: ~Node = cast::transmute(cur); + cur = next; + } + } + } +} + +#[cfg(test)] +mod test { + use prelude::*; + use super::queue; + use task; + + #[test] + fn smoke() { + let (mut c, mut p) = queue(0, ()); + p.push(1); + p.push(2); + assert_eq!(c.pop(), Some(1)); + assert_eq!(c.pop(), Some(2)); + assert_eq!(c.pop(), None); + p.push(3); + p.push(4); + assert_eq!(c.pop(), Some(3)); + assert_eq!(c.pop(), Some(4)); + assert_eq!(c.pop(), None); + } + + #[test] + fn drop_full() { + let (_, mut p) = queue(0, ()); + p.push(~1); + p.push(~2); + } + + #[test] + fn smoke_bound() { + let (mut c, mut p) = queue(1, ()); + p.push(1); + p.push(2); + assert_eq!(c.pop(), Some(1)); + assert_eq!(c.pop(), Some(2)); + assert_eq!(c.pop(), None); + p.push(3); + p.push(4); + assert_eq!(c.pop(), Some(3)); + assert_eq!(c.pop(), Some(4)); + assert_eq!(c.pop(), None); + } + + #[test] + fn stress() { + stress_bound(0); + stress_bound(1); + + fn stress_bound(bound: uint) { + let (c, mut p) = queue(bound, ()); + do task::spawn_sched(task::SingleThreaded) { + let mut c = c; + for _ in range(0, 100000) { + loop { + match c.pop() { + Some(1) => break, + Some(_) => fail!(), + None => {} + } + } + } + } + for _ in range(0, 100000) { + p.push(1); + } + } + } +} diff --git a/src/libstd/rt/task.rs b/src/libstd/rt/task.rs index 86cc895eb27..2adc32f33fb 100644 --- a/src/libstd/rt/task.rs +++ b/src/libstd/rt/task.rs @@ -26,7 +26,6 @@ use option::{Option, Some, None}; use rt::borrowck::BorrowRecord; use rt::borrowck; use rt::context::Context; -use rt::context; use rt::env; use io::Writer; use rt::kill::Death; -- cgit 1.4.1-3-g733a5