From 73729e94c87281dd7193dbdc86b4de2963b8fd72 Mon Sep 17 00:00:00 2001 From: Alex Crichton Date: Mon, 19 May 2014 17:50:57 -0700 Subject: [PATCH] std: Move comm primitives away from UnsafeArc They currently still use `&mut self`, this migration was aimed towards moving from UnsafeArc to Arc> --- src/libstd/comm/mod.rs | 49 +++++++++++++++++++---------------- src/libstd/comm/oneshot.rs | 2 +- src/libstd/sync/mpsc_queue.rs | 9 ++++--- src/libstd/sync/spsc_queue.rs | 9 ++++--- 4 files changed, 39 insertions(+), 30 deletions(-) diff --git a/src/libstd/comm/mod.rs b/src/libstd/comm/mod.rs index df0c6f3b8d3..fd5b92ba469 100644 --- a/src/libstd/comm/mod.rs +++ b/src/libstd/comm/mod.rs @@ -271,6 +271,8 @@ // And now that you've seen all the races that I found and attempted to fix, // here's the code for you to find some more! +use alloc::arc::Arc; + use cell::Cell; use clone::Clone; use iter::Iterator; @@ -283,7 +285,6 @@ use owned::Box; use result::{Ok, Err, Result}; use rt::local::Local; use rt::task::{Task, BlockedTask}; -use sync::arc::UnsafeArc; use ty::Unsafe; pub use comm::select::{Select, Handle}; @@ -352,7 +353,7 @@ pub struct Sender { /// The sending-half of Rust's synchronous channel type. This half can only be /// owned by one task, but it can be cloned to send to other tasks. pub struct SyncSender { - inner: UnsafeArc>, + inner: Arc>>, // can't share in an arc marker: marker::NoShare, } @@ -386,10 +387,10 @@ pub enum TrySendError { } enum Flavor { - Oneshot(UnsafeArc>), - Stream(UnsafeArc>), - Shared(UnsafeArc>), - Sync(UnsafeArc>), + Oneshot(Arc>>), + Stream(Arc>>), + Shared(Arc>>), + Sync(Arc>>), } #[doc(hidden)] @@ -435,8 +436,8 @@ impl UnsafeFlavor for Receiver { /// println!("{}", rx.recv()); /// ``` pub fn channel() -> (Sender, Receiver) { - let (a, b) = UnsafeArc::new2(oneshot::Packet::new()); - (Sender::new(Oneshot(b)), Receiver::new(Oneshot(a))) + let a = Arc::new(Unsafe::new(oneshot::Packet::new())); + (Sender::new(Oneshot(a.clone())), Receiver::new(Oneshot(a))) } /// Creates a new synchronous, bounded channel. @@ -471,8 +472,8 @@ pub fn channel() -> (Sender, Receiver) { /// assert_eq!(rx.recv(), 2); /// ``` pub fn sync_channel(bound: uint) -> (SyncSender, Receiver) { - let (a, b) = UnsafeArc::new2(sync::Packet::new(bound)); - (SyncSender::new(a), Receiver::new(Sync(b))) + let a = Arc::new(Unsafe::new(sync::Packet::new(bound))); + (SyncSender::new(a.clone()), Receiver::new(Sync(a))) } //////////////////////////////////////////////////////////////////////////////// @@ -557,13 +558,13 @@ impl Sender { let (new_inner, ret) = match *unsafe { self.inner() } { Oneshot(ref p) => { - let p = p.get(); unsafe { + let p = p.get(); if !(*p).sent() { return (*p).send(t); } else { - let (a, b) = UnsafeArc::new2(stream::Packet::new()); - match (*p).upgrade(Receiver::new(Stream(b))) { + let a = Arc::new(Unsafe::new(stream::Packet::new())); + match (*p).upgrade(Receiver::new(Stream(a.clone()))) { oneshot::UpSuccess => { let ret = (*a.get()).send(t); (a, ret) @@ -598,17 +599,21 @@ impl Clone for Sender { fn clone(&self) -> Sender { let (packet, sleeper) = match *unsafe { self.inner() } { Oneshot(ref p) => { - let (a, b) = UnsafeArc::new2(shared::Packet::new()); - match unsafe { (*p.get()).upgrade(Receiver::new(Shared(a))) } { - oneshot::UpSuccess | oneshot::UpDisconnected => (b, None), - oneshot::UpWoke(task) => (b, Some(task)) + let a = Arc::new(Unsafe::new(shared::Packet::new())); + match unsafe { + (*p.get()).upgrade(Receiver::new(Shared(a.clone()))) + } { + oneshot::UpSuccess | oneshot::UpDisconnected => (a, None), + oneshot::UpWoke(task) => (a, Some(task)) } } Stream(ref p) => { - let (a, b) = UnsafeArc::new2(shared::Packet::new()); - match unsafe { (*p.get()).upgrade(Receiver::new(Shared(a))) } { - stream::UpSuccess | stream::UpDisconnected => (b, None), - stream::UpWoke(task) => (b, Some(task)), + let a = Arc::new(Unsafe::new(shared::Packet::new())); + match unsafe { + (*p.get()).upgrade(Receiver::new(Shared(a.clone()))) + } { + stream::UpSuccess | stream::UpDisconnected => (a, None), + stream::UpWoke(task) => (a, Some(task)), } } Shared(ref p) => { @@ -645,7 +650,7 @@ impl Drop for Sender { //////////////////////////////////////////////////////////////////////////////// impl SyncSender { - fn new(inner: UnsafeArc>) -> SyncSender { + fn new(inner: Arc>>) -> SyncSender { SyncSender { inner: inner, marker: marker::NoShare } } diff --git a/src/libstd/comm/oneshot.rs b/src/libstd/comm/oneshot.rs index a7124e50b66..f9e8fd1e534 100644 --- a/src/libstd/comm/oneshot.rs +++ b/src/libstd/comm/oneshot.rs @@ -15,7 +15,7 @@ /// this type is to have one and exactly one allocation when the chan/port pair /// is created. /// -/// Another possible optimization would be to not use an UnsafeArc box because +/// Another possible optimization would be to not use an Arc box because /// in theory we know when the shared packet can be deallocated (no real need /// for the atomic reference counting), but I was having trouble how to destroy /// the data early in a drop of a Port. diff --git a/src/libstd/sync/mpsc_queue.rs b/src/libstd/sync/mpsc_queue.rs index 23afb9487ec..f2f95da1842 100644 --- a/src/libstd/sync/mpsc_queue.rs +++ b/src/libstd/sync/mpsc_queue.rs @@ -158,9 +158,10 @@ impl Drop for Queue { mod tests { use prelude::*; + use alloc::arc::Arc; + use native; use super::{Queue, Data, Empty, Inconsistent}; - use sync::arc::UnsafeArc; #[test] fn test_full() { @@ -179,14 +180,14 @@ mod tests { Inconsistent | Data(..) => fail!() } let (tx, rx) = channel(); - let q = UnsafeArc::new(q); + let q = Arc::new(q); for _ in range(0, nthreads) { let tx = tx.clone(); let q = q.clone(); native::task::spawn(proc() { for i in range(0, nmsgs) { - unsafe { (*q.get()).push(i); } + q.push(i); } tx.send(()); }); @@ -194,7 +195,7 @@ mod tests { let mut i = 0u; while i < nthreads * nmsgs { - match unsafe { (*q.get()).pop() } { + match q.pop() { Empty | Inconsistent => {}, Data(_) => { i += 1 } } diff --git a/src/libstd/sync/spsc_queue.rs b/src/libstd/sync/spsc_queue.rs index b9827ee6b2a..093933c82fc 100644 --- a/src/libstd/sync/spsc_queue.rs +++ b/src/libstd/sync/spsc_queue.rs @@ -52,7 +52,7 @@ struct Node { } /// The single-producer single-consumer queue. This structure is not cloneable, -/// but it can be safely shared in an UnsafeArc if it is guaranteed that there +/// but it can be safely shared in an Arc if it is guaranteed that there /// is only one popper and one pusher touching the queue at any one point in /// time. pub struct Queue { @@ -227,9 +227,11 @@ impl Drop for Queue { #[cfg(test)] mod test { use prelude::*; + + use alloc::arc::Arc; use native; + use super::Queue; - use sync::arc::UnsafeArc; #[test] fn smoke() { @@ -274,7 +276,8 @@ mod test { stress_bound(1); fn stress_bound(bound: uint) { - let (a, b) = UnsafeArc::new2(Queue::new(bound)); + let a = Arc::new(Queue::new(bound)); + let b = a.clone(); let (tx, rx) = channel(); native::task::spawn(proc() { for _ in range(0, 100000) {