diff options
| author | bors <bors@rust-lang.org> | 2013-12-10 19:16:19 -0800 |
|---|---|---|
| committer | bors <bors@rust-lang.org> | 2013-12-10 19:16:19 -0800 |
| commit | b8b16ae0996074861693f0f76d5d937fafe6a37e (patch) | |
| tree | 03f3fadcb4558ec1ddda3d61669636c1fcad1264 /src/libstd | |
| parent | ac4dd9efee9248f4c3235460f3c93acc9932eb5a (diff) | |
| parent | fd7a513bef7fe3c6f5128cc53135facca37f23e5 (diff) | |
| download | rust-b8b16ae0996074861693f0f76d5d937fafe6a37e.tar.gz rust-b8b16ae0996074861693f0f76d5d937fafe6a37e.zip | |
auto merge of #10791 : pcwalton/rust/decelling, r=pcwalton
34 uses of `Cell` remain. r? @alexcrichton
Diffstat (limited to 'src/libstd')
37 files changed, 683 insertions, 741 deletions
diff --git a/src/libstd/at_vec.rs b/src/libstd/at_vec.rs index 2b105a3fa7d..2b64c5c83fb 100644 --- a/src/libstd/at_vec.rs +++ b/src/libstd/at_vec.rs @@ -169,6 +169,7 @@ pub mod raw { use at_vec::capacity; use cast; use cast::{transmute, transmute_copy}; + use option::None; use ptr; use mem; use uint; @@ -259,9 +260,8 @@ pub mod raw { use rt::local::Local; use rt::task::Task; - Local::borrow(|task: &mut Task| { - task.heap.realloc(ptr as *mut Box<()>, size) as *() - }) + let mut task = Local::borrow(None::<Task>); + task.get().heap.realloc(ptr as *mut Box<()>, size) as *() } } diff --git a/src/libstd/cell.rs b/src/libstd/cell.rs index e49cf3e5303..280c79f1fd4 100644 --- a/src/libstd/cell.rs +++ b/src/libstd/cell.rs @@ -14,66 +14,6 @@ use prelude::*; use cast; use util::NonCopyable; - -/* -A dynamic, mutable location. - -Similar to a mutable option type, but friendlier. -*/ - -#[no_freeze] -#[deriving(Clone, DeepClone, Eq)] -#[allow(missing_doc)] -pub struct Cell<T> { - priv value: Option<T> -} - -impl<T> Cell<T> { - /// Creates a new full cell with the given value. - pub fn new(value: T) -> Cell<T> { - Cell { value: Some(value) } - } - - /// Yields the value, failing if the cell is empty. - pub fn take(&self) -> T { - let this = unsafe { cast::transmute_mut(self) }; - if this.is_empty() { - fail!("attempt to take an empty cell"); - } - - this.value.take_unwrap() - } - - /// Yields the value if the cell is full, or `None` if it is empty. - pub fn take_opt(&self) -> Option<T> { - let this = unsafe { cast::transmute_mut(self) }; - this.value.take() - } - - /// Returns true if the cell is empty and false if the cell is full. - pub fn is_empty(&self) -> bool { - self.value.is_none() - } -} - -#[test] -fn test_basic() { - let value_cell = Cell::new(~10); - assert!(!value_cell.is_empty()); - let value = value_cell.take(); - assert!(value == ~10); - assert!(value_cell.is_empty()); -} - -#[test] -#[should_fail] -fn test_take_empty() { - let value_cell: Cell<~int> = Cell::new(~0); - value_cell.take(); - value_cell.take(); -} - - /// A mutable memory location with dynamically checked borrow rules #[no_freeze] pub struct RefCell<T> { diff --git a/src/libstd/condition.rs b/src/libstd/condition.rs index e34e94ac10c..c5d6ce2f3df 100644 --- a/src/libstd/condition.rs +++ b/src/libstd/condition.rs @@ -162,7 +162,7 @@ impl<T, U> Condition<T, U> { /// /// Normally this object is not dealt with directly, but rather it's directly /// used after being returned from `trap` -struct Trap<'self, T, U> { +pub struct Trap<'self, T, U> { priv cond: &'self Condition<T, U>, priv handler: @Handler<T, U> } @@ -187,10 +187,24 @@ impl<'self, T, U> Trap<'self, T, U> { local_data::set(self.cond.key, self.handler); inner() } + + /// Returns a guard that will automatically reset the condition upon + /// exit of the scope. This is useful if you want to use conditions with + /// an RAII pattern. + pub fn guard(&self) -> Guard<'self,T,U> { + let guard = Guard { + cond: self.cond + }; + debug!("Guard: pushing handler to TLS"); + local_data::set(self.cond.key, self.handler); + guard + } } -#[doc(hidden)] -struct Guard<'self, T, U> { +/// A guard that will automatically reset the condition handler upon exit of +/// the scope. This is useful if you want to use conditions with an RAII +/// pattern. +pub struct Guard<'self, T, U> { priv cond: &'self Condition<T, U> } diff --git a/src/libstd/io/fs.rs b/src/libstd/io/fs.rs index 23bf5cdc564..5bfdf685bea 100644 --- a/src/libstd/io/fs.rs +++ b/src/libstd/io/fs.rs @@ -51,7 +51,7 @@ use iter::Iterator; use super::{Reader, Writer, Seek}; use super::{SeekStyle, Read, Write, Open, IoError, Truncate, FileMode, FileAccess, FileStat, io_error, FilePermission}; -use rt::rtio::{RtioFileStream, IoFactory, with_local_io}; +use rt::rtio::{RtioFileStream, IoFactory, LocalIo}; use io; use option::{Some, None, Option}; use result::{Ok, Err, Result}; @@ -76,15 +76,14 @@ pub struct File { } fn io_raise<T>(f: |io: &mut IoFactory| -> Result<T, IoError>) -> Option<T> { - with_local_io(|io| { - match f(io) { - Ok(t) => Some(t), - Err(ioerr) => { - io_error::cond.raise(ioerr); - None - } + let mut io = LocalIo::borrow(); + match f(io.get()) { + Ok(t) => Some(t), + Err(ioerr) => { + io_error::cond.raise(ioerr); + None } - }) + } } impl File { @@ -132,19 +131,18 @@ impl File { pub fn open_mode(path: &Path, mode: FileMode, access: FileAccess) -> Option<File> { - with_local_io(|io| { - match io.fs_open(&path.to_c_str(), mode, access) { - Ok(fd) => Some(File { - path: path.clone(), - fd: fd, - last_nread: -1 - }), - Err(ioerr) => { - io_error::cond.raise(ioerr); - None - } + let mut io = LocalIo::borrow(); + match io.get().fs_open(&path.to_c_str(), mode, access) { + Ok(fd) => Some(File { + path: path.clone(), + fd: fd, + last_nread: -1 + }), + Err(ioerr) => { + io_error::cond.raise(ioerr); + None } - }) + } } /// Attempts to open a file in read-only mode. This function is equivalent to diff --git a/src/libstd/io/mod.rs b/src/libstd/io/mod.rs index edc6728a0c1..d5e216e2426 100644 --- a/src/libstd/io/mod.rs +++ b/src/libstd/io/mod.rs @@ -241,6 +241,7 @@ Out of scope #[allow(missing_doc)]; use cast; +use condition::Guard; use container::Container; use int; use iter::Iterator; @@ -394,12 +395,12 @@ condition! { /// Helper for wrapper calls where you want to /// ignore any io_errors that might be raised -pub fn ignore_io_error<T>(cb: || -> T) -> T { +pub fn ignore_io_error() -> Guard<'static,IoError,()> { io_error::cond.trap(|_| { // just swallow the error.. downstream users // who can make a decision based on a None result // won't care - }).inside(|| cb()) + }).guard() } /// Helper for catching an I/O error and wrapping it in a Result object. The diff --git a/src/libstd/io/native/mod.rs b/src/libstd/io/native/mod.rs index 00b26116e67..d9dccc84f1c 100644 --- a/src/libstd/io/native/mod.rs +++ b/src/libstd/io/native/mod.rs @@ -223,3 +223,6 @@ impl rtio::IoFactory for IoFactory { Err(unimpl()) } } + +pub static mut NATIVE_IO_FACTORY: IoFactory = IoFactory; + diff --git a/src/libstd/io/net/addrinfo.rs b/src/libstd/io/net/addrinfo.rs index 03af64cc6dc..7df4fdd2266 100644 --- a/src/libstd/io/net/addrinfo.rs +++ b/src/libstd/io/net/addrinfo.rs @@ -21,7 +21,7 @@ use option::{Option, Some, None}; use result::{Ok, Err}; use io::{io_error}; use io::net::ip::{SocketAddr, IpAddr}; -use rt::rtio::{IoFactory, with_local_io}; +use rt::rtio::{IoFactory, LocalIo}; use vec::ImmutableVector; /// Hints to the types of sockets that are desired when looking up hosts @@ -95,17 +95,16 @@ pub fn get_host_addresses(host: &str) -> Option<~[IpAddr]> { /// /// XXX: this is not public because the `Hint` structure is not ready for public /// consumption just yet. -fn lookup(hostname: Option<&str>, servname: Option<&str>, - hint: Option<Hint>) -> Option<~[Info]> { - with_local_io(|io| { - match io.get_host_addresses(hostname, servname, hint) { - Ok(i) => Some(i), - Err(ioerr) => { - io_error::cond.raise(ioerr); - None - } +fn lookup(hostname: Option<&str>, servname: Option<&str>, hint: Option<Hint>) + -> Option<~[Info]> { + let mut io = LocalIo::borrow(); + match io.get().get_host_addresses(hostname, servname, hint) { + Ok(i) => Some(i), + Err(ioerr) => { + io_error::cond.raise(ioerr); + None } - }) + } } #[cfg(test)] diff --git a/src/libstd/io/net/tcp.rs b/src/libstd/io/net/tcp.rs index aa7a64d2210..3c7582db7b8 100644 --- a/src/libstd/io/net/tcp.rs +++ b/src/libstd/io/net/tcp.rs @@ -13,8 +13,8 @@ use result::{Ok, Err}; use io::net::ip::SocketAddr; use io::{Reader, Writer, Listener, Acceptor}; use io::{io_error, EndOfFile}; -use rt::rtio::{IoFactory, with_local_io, - RtioSocket, RtioTcpListener, RtioTcpAcceptor, RtioTcpStream}; +use rt::rtio::{IoFactory, LocalIo, RtioSocket, RtioTcpListener}; +use rt::rtio::{RtioTcpAcceptor, RtioTcpStream}; pub struct TcpStream { priv obj: ~RtioTcpStream @@ -26,15 +26,17 @@ impl TcpStream { } pub fn connect(addr: SocketAddr) -> Option<TcpStream> { - with_local_io(|io| { - match io.tcp_connect(addr) { - Ok(s) => Some(TcpStream::new(s)), - Err(ioerr) => { - io_error::cond.raise(ioerr); - None - } + let result = { + let mut io = LocalIo::borrow(); + io.get().tcp_connect(addr) + }; + match result { + Ok(s) => Some(TcpStream::new(s)), + Err(ioerr) => { + io_error::cond.raise(ioerr); + None } - }) + } } pub fn peer_name(&mut self) -> Option<SocketAddr> { @@ -92,15 +94,14 @@ pub struct TcpListener { impl TcpListener { pub fn bind(addr: SocketAddr) -> Option<TcpListener> { - with_local_io(|io| { - match io.tcp_bind(addr) { - Ok(l) => Some(TcpListener { obj: l }), - Err(ioerr) => { - io_error::cond.raise(ioerr); - None - } + let mut io = LocalIo::borrow(); + match io.get().tcp_bind(addr) { + Ok(l) => Some(TcpListener { obj: l }), + Err(ioerr) => { + io_error::cond.raise(ioerr); + None } - }) + } } pub fn socket_name(&mut self) -> Option<SocketAddr> { @@ -146,7 +147,6 @@ impl Acceptor<TcpStream> for TcpAcceptor { #[cfg(test)] mod test { use super::*; - use cell::Cell; use rt::test::*; use io::net::ip::{Ipv4Addr, SocketAddr}; use io::*; @@ -196,12 +196,10 @@ mod test { do run_in_mt_newsched_task { let addr = next_test_ip4(); let (port, chan) = oneshot(); - let port = Cell::new(port); - let chan = Cell::new(chan); do spawntask { let mut acceptor = TcpListener::bind(addr).listen(); - chan.take().send(()); + chan.send(()); let mut stream = acceptor.accept(); let mut buf = [0]; stream.read(buf); @@ -209,7 +207,7 @@ mod test { } do spawntask { - port.take().recv(); + port.recv(); let mut stream = TcpStream::connect(addr); stream.write([99]); } @@ -221,12 +219,10 @@ mod test { do run_in_mt_newsched_task { let addr = next_test_ip6(); let (port, chan) = oneshot(); - let port = Cell::new(port); - let chan = Cell::new(chan); do spawntask { let mut acceptor = TcpListener::bind(addr).listen(); - chan.take().send(()); + chan.send(()); let mut stream = acceptor.accept(); let mut buf = [0]; stream.read(buf); @@ -234,7 +230,7 @@ mod test { } do spawntask { - port.take().recv(); + port.recv(); let mut stream = TcpStream::connect(addr); stream.write([99]); } @@ -246,12 +242,10 @@ mod test { do run_in_mt_newsched_task { let addr = next_test_ip4(); let (port, chan) = oneshot(); - let port = Cell::new(port); - let chan = Cell::new(chan); do spawntask { let mut acceptor = TcpListener::bind(addr).listen(); - chan.take().send(()); + chan.send(()); let mut stream = acceptor.accept(); let mut buf = [0]; let nread = stream.read(buf); @@ -259,7 +253,7 @@ mod test { } do spawntask { - port.take().recv(); + port.recv(); let _stream = TcpStream::connect(addr); // Close } @@ -271,12 +265,10 @@ mod test { do run_in_mt_newsched_task { let addr = next_test_ip6(); let (port, chan) = oneshot(); - let port = Cell::new(port); - let chan = Cell::new(chan); do spawntask { let mut acceptor = TcpListener::bind(addr).listen(); - chan.take().send(()); + chan.send(()); let mut stream = acceptor.accept(); let mut buf = [0]; let nread = stream.read(buf); @@ -284,7 +276,7 @@ mod test { } do spawntask { - port.take().recv(); + port.recv(); let _stream = TcpStream::connect(addr); // Close } @@ -296,12 +288,10 @@ mod test { do run_in_mt_newsched_task { let addr = next_test_ip4(); let (port, chan) = oneshot(); - let port = Cell::new(port); - let chan = Cell::new(chan); do spawntask { let mut acceptor = TcpListener::bind(addr).listen(); - chan.take().send(()); + chan.send(()); let mut stream = acceptor.accept(); let mut buf = [0]; let nread = stream.read(buf); @@ -319,7 +309,7 @@ mod test { } do spawntask { - port.take().recv(); + port.recv(); let _stream = TcpStream::connect(addr); // Close } @@ -331,12 +321,10 @@ mod test { do run_in_mt_newsched_task { let addr = next_test_ip6(); let (port, chan) = oneshot(); - let port = Cell::new(port); - let chan = Cell::new(chan); do spawntask { let mut acceptor = TcpListener::bind(addr).listen(); - chan.take().send(()); + chan.send(()); let mut stream = acceptor.accept(); let mut buf = [0]; let nread = stream.read(buf); @@ -354,7 +342,7 @@ mod test { } do spawntask { - port.take().recv(); + port.recv(); let _stream = TcpStream::connect(addr); // Close } @@ -366,12 +354,10 @@ mod test { do run_in_mt_newsched_task { let addr = next_test_ip4(); let (port, chan) = oneshot(); - let port = Cell::new(port); - let chan = Cell::new(chan); do spawntask { let mut acceptor = TcpListener::bind(addr).listen(); - chan.take().send(()); + chan.send(()); let mut stream = acceptor.accept(); let buf = [0]; loop { @@ -392,7 +378,7 @@ mod test { } do spawntask { - port.take().recv(); + port.recv(); let _stream = TcpStream::connect(addr); // Close } @@ -404,12 +390,10 @@ mod test { do run_in_mt_newsched_task { let addr = next_test_ip6(); let (port, chan) = oneshot(); - let port = Cell::new(port); - let chan = Cell::new(chan); do spawntask { let mut acceptor = TcpListener::bind(addr).listen(); - chan.take().send(()); + chan.send(()); let mut stream = acceptor.accept(); let buf = [0]; loop { @@ -430,7 +414,7 @@ mod test { } do spawntask { - port.take().recv(); + port.recv(); let _stream = TcpStream::connect(addr); // Close } @@ -443,12 +427,10 @@ mod test { let addr = next_test_ip4(); let max = 10; let (port, chan) = oneshot(); - let port = Cell::new(port); - let chan = Cell::new(chan); do spawntask { let mut acceptor = TcpListener::bind(addr).listen(); - chan.take().send(()); + chan.send(()); for ref mut stream in acceptor.incoming().take(max) { let mut buf = [0]; stream.read(buf); @@ -457,7 +439,7 @@ mod test { } do spawntask { - port.take().recv(); + port.recv(); max.times(|| { let mut stream = TcpStream::connect(addr); stream.write([99]); @@ -472,12 +454,10 @@ mod test { let addr = next_test_ip6(); let max = 10; let (port, chan) = oneshot(); - let port = Cell::new(port); - let chan = Cell::new(chan); do spawntask { let mut acceptor = TcpListener::bind(addr).listen(); - chan.take().send(()); + chan.send(()); for ref mut stream in acceptor.incoming().take(max) { let mut buf = [0]; stream.read(buf); @@ -486,7 +466,7 @@ mod test { } do spawntask { - port.take().recv(); + port.recv(); max.times(|| { let mut stream = TcpStream::connect(addr); stream.write([99]); @@ -501,16 +481,14 @@ mod test { let addr = next_test_ip4(); static MAX: int = 10; let (port, chan) = oneshot(); - let chan = Cell::new(chan); do spawntask { let mut acceptor = TcpListener::bind(addr).listen(); - chan.take().send(()); + chan.send(()); for (i, stream) in acceptor.incoming().enumerate().take(MAX as uint) { - let stream = Cell::new(stream); // Start another task to handle the connection do spawntask { - let mut stream = stream.take(); + let mut stream = stream; let mut buf = [0]; stream.read(buf); assert!(buf[0] == i as u8); @@ -543,16 +521,14 @@ mod test { let addr = next_test_ip6(); static MAX: int = 10; let (port, chan) = oneshot(); - let chan = Cell::new(chan); do spawntask { let mut acceptor = TcpListener::bind(addr).listen(); - chan.take().send(()); + chan.send(()); for (i, stream) in acceptor.incoming().enumerate().take(MAX as uint) { - let stream = Cell::new(stream); // Start another task to handle the connection do spawntask { - let mut stream = stream.take(); + let mut stream = stream; let mut buf = [0]; stream.read(buf); assert!(buf[0] == i as u8); @@ -585,16 +561,14 @@ mod test { let addr = next_test_ip4(); static MAX: int = 10; let (port, chan) = oneshot(); - let chan = Cell::new(chan); do spawntask { let mut acceptor = TcpListener::bind(addr).listen(); - chan.take().send(()); + chan.send(()); for stream in acceptor.incoming().take(MAX as uint) { - let stream = Cell::new(stream); // Start another task to handle the connection do spawntask_later { - let mut stream = stream.take(); + let mut stream = stream; let mut buf = [0]; stream.read(buf); assert!(buf[0] == 99); @@ -626,16 +600,14 @@ mod test { let addr = next_test_ip6(); static MAX: int = 10; let (port, chan) = oneshot(); - let chan = Cell::new(chan); do spawntask { let mut acceptor = TcpListener::bind(addr).listen(); - chan.take().send(()); + chan.send(()); for stream in acceptor.incoming().take(MAX as uint) { - let stream = Cell::new(stream); // Start another task to handle the connection do spawntask_later { - let mut stream = stream.take(); + let mut stream = stream; let mut buf = [0]; stream.read(buf); assert!(buf[0] == 99); @@ -682,18 +654,16 @@ mod test { fn peer_name(addr: SocketAddr) { do run_in_mt_newsched_task { let (port, chan) = oneshot(); - let port = Cell::new(port); - let chan = Cell::new(chan); do spawntask { let mut acceptor = TcpListener::bind(addr).listen(); - chan.take().send(()); + chan.send(()); acceptor.accept(); } do spawntask { - port.take().recv(); + port.recv(); let stream = TcpStream::connect(addr); assert!(stream.is_some()); diff --git a/src/libstd/io/net/udp.rs b/src/libstd/io/net/udp.rs index f02fc1ae447..87cf59aba3b 100644 --- a/src/libstd/io/net/udp.rs +++ b/src/libstd/io/net/udp.rs @@ -13,7 +13,7 @@ use result::{Ok, Err}; use io::net::ip::SocketAddr; use io::{Reader, Writer}; use io::{io_error, EndOfFile}; -use rt::rtio::{RtioSocket, RtioUdpSocket, IoFactory, with_local_io}; +use rt::rtio::{RtioSocket, RtioUdpSocket, IoFactory, LocalIo}; pub struct UdpSocket { priv obj: ~RtioUdpSocket @@ -21,15 +21,14 @@ pub struct UdpSocket { impl UdpSocket { pub fn bind(addr: SocketAddr) -> Option<UdpSocket> { - with_local_io(|io| { - match io.udp_bind(addr) { - Ok(s) => Some(UdpSocket { obj: s }), - Err(ioerr) => { - io_error::cond.raise(ioerr); - None - } + let mut io = LocalIo::borrow(); + match io.get().udp_bind(addr) { + Ok(s) => Some(UdpSocket { obj: s }), + Err(ioerr) => { + io_error::cond.raise(ioerr); + None } - }) + } } pub fn recvfrom(&mut self, buf: &mut [u8]) -> Option<(uint, SocketAddr)> { @@ -110,7 +109,6 @@ mod test { use io::*; use option::{Some, None}; use rt::comm::oneshot; - use cell::Cell; #[test] #[ignore] fn bind_error() { @@ -134,13 +132,11 @@ mod test { let server_ip = next_test_ip4(); let client_ip = next_test_ip4(); let (port, chan) = oneshot(); - let port = Cell::new(port); - let chan = Cell::new(chan); do spawntask { match UdpSocket::bind(server_ip) { Some(ref mut server) => { - chan.take().send(()); + chan.send(()); let mut buf = [0]; match server.recvfrom(buf) { Some((nread, src)) => { @@ -158,7 +154,7 @@ mod test { do spawntask { match UdpSocket::bind(client_ip) { Some(ref mut client) => { - port.take().recv(); + port.recv(); client.sendto([99], server_ip) } None => fail!() @@ -173,13 +169,11 @@ mod test { let server_ip = next_test_ip6(); let client_ip = next_test_ip6(); let (port, chan) = oneshot(); - let port = Cell::new(port); - let chan = Cell::new(chan); do spawntask { match UdpSocket::bind(server_ip) { Some(ref mut server) => { - chan.take().send(()); + chan.send(()); let mut buf = [0]; match server.recvfrom(buf) { Some((nread, src)) => { @@ -197,7 +191,7 @@ mod test { do spawntask { match UdpSocket::bind(client_ip) { Some(ref mut client) => { - port.take().recv(); + port.recv(); client.sendto([99], server_ip) } None => fail!() @@ -212,15 +206,13 @@ mod test { let server_ip = next_test_ip4(); let client_ip = next_test_ip4(); let (port, chan) = oneshot(); - let port = Cell::new(port); - let chan = Cell::new(chan); do spawntask { match UdpSocket::bind(server_ip) { Some(server) => { let server = ~server; let mut stream = server.connect(client_ip); - chan.take().send(()); + chan.send(()); let mut buf = [0]; match stream.read(buf) { Some(nread) => { @@ -239,7 +231,7 @@ mod test { Some(client) => { let client = ~client; let mut stream = client.connect(server_ip); - port.take().recv(); + port.recv(); stream.write([99]); } None => fail!() @@ -254,15 +246,13 @@ mod test { let server_ip = next_test_ip6(); let client_ip = next_test_ip6(); let (port, chan) = oneshot(); - let port = Cell::new(port); - let chan = Cell::new(chan); do spawntask { match UdpSocket::bind(server_ip) { Some(server) => { let server = ~server; let mut stream = server.connect(client_ip); - chan.take().send(()); + chan.send(()); let mut buf = [0]; match stream.read(buf) { Some(nread) => { @@ -281,7 +271,7 @@ mod test { Some(client) => { let client = ~client; let mut stream = client.connect(server_ip); - port.take().recv(); + port.recv(); stream.write([99]); } None => fail!() diff --git a/src/libstd/io/net/unix.rs b/src/libstd/io/net/unix.rs index 6d2deccaa4c..c1f75465d9c 100644 --- a/src/libstd/io/net/unix.rs +++ b/src/libstd/io/net/unix.rs @@ -25,7 +25,7 @@ instances as clients. use prelude::*; use c_str::ToCStr; -use rt::rtio::{IoFactory, RtioUnixListener, with_local_io}; +use rt::rtio::{IoFactory, LocalIo, RtioUnixListener}; use rt::rtio::{RtioUnixAcceptor, RtioPipe}; use io::pipe::PipeStream; use io::{io_error, Listener, Acceptor, Reader, Writer}; @@ -59,15 +59,14 @@ impl UnixStream { /// stream.write([1, 2, 3]); /// pub fn connect<P: ToCStr>(path: &P) -> Option<UnixStream> { - with_local_io(|io| { - match io.unix_connect(&path.to_c_str()) { - Ok(s) => Some(UnixStream::new(s)), - Err(ioerr) => { - io_error::cond.raise(ioerr); - None - } + let mut io = LocalIo::borrow(); + match io.get().unix_connect(&path.to_c_str()) { + Ok(s) => Some(UnixStream::new(s)), + Err(ioerr) => { + io_error::cond.raise(ioerr); + None } - }) + } } } @@ -108,15 +107,14 @@ impl UnixListener { /// } /// pub fn bind<P: ToCStr>(path: &P) -> Option<UnixListener> { - with_local_io(|io| { - match io.unix_bind(&path.to_c_str()) { - Ok(s) => Some(UnixListener{ obj: s }), - Err(ioerr) => { - io_error::cond.raise(ioerr); - None - } + let mut io = LocalIo::borrow(); + match io.get().unix_bind(&path.to_c_str()) { + Ok(s) => Some(UnixListener{ obj: s }), + Err(ioerr) => { + io_error::cond.raise(ioerr); + None } - }) + } } } @@ -152,32 +150,26 @@ impl Acceptor<UnixStream> for UnixAcceptor { mod tests { use prelude::*; use super::*; - use cell::Cell; use rt::test::*; use io::*; use rt::comm::oneshot; fn smalltest(server: proc(UnixStream), client: proc(UnixStream)) { - let server = Cell::new(server); - let client = Cell::new(client); do run_in_mt_newsched_task { - let server = Cell::new(server.take()); - let client = Cell::new(client.take()); let path1 = next_test_unix(); let path2 = path1.clone(); let (port, chan) = oneshot(); - let port = Cell::new(port); - let chan = Cell::new(chan); + let (client, server) = (client, server); do spawntask { let mut acceptor = UnixListener::bind(&path1).listen(); - chan.take().send(()); - server.take()(acceptor.accept().unwrap()); + chan.send(()); + server(acceptor.accept().unwrap()); } do spawntask { - port.take().recv(); - client.take()(UnixStream::connect(&path2).unwrap()); + port.recv(); + client(UnixStream::connect(&path2).unwrap()); } } } @@ -260,12 +252,10 @@ mod tests { let path1 = next_test_unix(); let path2 = path1.clone(); let (port, chan) = oneshot(); - let port = Cell::new(port); - let chan = Cell::new(chan); do spawntask { let mut acceptor = UnixListener::bind(&path1).listen(); - chan.take().send(()); + chan.send(()); times.times(|| { let mut client = acceptor.accept(); let mut buf = [0]; @@ -275,7 +265,7 @@ mod tests { } do spawntask { - port.take().recv(); + port.recv(); times.times(|| { let mut stream = UnixStream::connect(&path2); stream.write([100]); diff --git a/src/libstd/io/pipe.rs b/src/libstd/io/pipe.rs index 373de1649ed..252575ee445 100644 --- a/src/libstd/io/pipe.rs +++ b/src/libstd/io/pipe.rs @@ -17,7 +17,7 @@ use prelude::*; use super::{Reader, Writer}; use io::{io_error, EndOfFile}; use io::native::file; -use rt::rtio::{RtioPipe, with_local_io}; +use rt::rtio::{LocalIo, RtioPipe}; pub struct PipeStream { priv obj: ~RtioPipe, @@ -44,15 +44,14 @@ impl PipeStream { /// If the pipe cannot be created, an error will be raised on the /// `io_error` condition. pub fn open(fd: file::fd_t) -> Option<PipeStream> { - with_local_io(|io| { - match io.pipe_open(fd) { - Ok(obj) => Some(PipeStream { obj: obj }), - Err(e) => { - io_error::cond.raise(e); - None - } + let mut io = LocalIo::borrow(); + match io.get().pipe_open(fd) { + Ok(obj) => Some(PipeStream { obj: obj }), + Err(e) => { + io_error::cond.raise(e); + None } - }) + } } pub fn new(inner: ~RtioPipe) -> PipeStream { diff --git a/src/libstd/io/process.rs b/src/libstd/io/process.rs index 308e969c43a..1c86ac84bbb 100644 --- a/src/libstd/io/process.rs +++ b/src/libstd/io/process.rs @@ -11,12 +11,11 @@ //! Bindings for executing child processes use prelude::*; -use cell::Cell; use libc; use io; use io::io_error; -use rt::rtio::{RtioProcess, IoFactory, with_local_io}; +use rt::rtio::{RtioProcess, IoFactory, LocalIo}; use fmt; @@ -120,21 +119,19 @@ impl Process { /// Creates a new pipe initialized, but not bound to any particular /// source/destination pub fn new(config: ProcessConfig) -> Option<Process> { - let config = Cell::new(config); - with_local_io(|io| { - match io.spawn(config.take()) { - Ok((p, io)) => Some(Process{ - handle: p, - io: io.move_iter().map(|p| - p.map(|p| io::PipeStream::new(p)) - ).collect() - }), - Err(ioerr) => { - io_error::cond.raise(ioerr); - None - } + let mut io = LocalIo::borrow(); + match io.get().spawn(config) { + Ok((p, io)) => Some(Process{ + handle: p, + io: io.move_iter().map(|p| + p.map(|p| io::PipeStream::new(p)) + ).collect() + }), + Err(ioerr) => { + io_error::cond.raise(ioerr); + None } - }) + } } /// Returns the process id of this child process diff --git a/src/libstd/io/signal.rs b/src/libstd/io/signal.rs index f6e79a03323..43659d24c35 100644 --- a/src/libstd/io/signal.rs +++ b/src/libstd/io/signal.rs @@ -24,9 +24,8 @@ use comm::{Port, SharedChan, stream}; use container::{Map, MutableMap}; use hashmap; use io::io_error; -use option::{Some, None}; use result::{Err, Ok}; -use rt::rtio::{IoFactory, RtioSignal, with_local_io}; +use rt::rtio::{IoFactory, LocalIo, RtioSignal}; #[repr(int)] #[deriving(Eq, IterBytes)] @@ -123,18 +122,17 @@ impl Listener { if self.handles.contains_key(&signum) { return true; // self is already listening to signum, so succeed } - with_local_io(|io| { - match io.signal(signum, self.chan.clone()) { - Ok(w) => { - self.handles.insert(signum, w); - Some(()) - }, - Err(ioerr) => { - io_error::cond.raise(ioerr); - None - } + let mut io = LocalIo::borrow(); + match io.get().signal(signum, self.chan.clone()) { + Ok(w) => { + self.handles.insert(signum, w); + true + }, + Err(ioerr) => { + io_error::cond.raise(ioerr); + false } - }).is_some() + } } /// Unregisters a signal. If this listener currently had a handler diff --git a/src/libstd/io/stdio.rs b/src/libstd/io/stdio.rs index 466a65a227e..41337075aa9 100644 --- a/src/libstd/io/stdio.rs +++ b/src/libstd/io/stdio.rs @@ -31,7 +31,7 @@ use libc; use option::{Option, Some, None}; use result::{Ok, Err}; use io::buffered::LineBufferedWriter; -use rt::rtio::{IoFactory, RtioTTY, RtioFileStream, with_local_io, DontClose}; +use rt::rtio::{DontClose, IoFactory, LocalIo, RtioFileStream, RtioTTY}; use super::{Reader, Writer, io_error, IoError, OtherIoError, standard_error, EndOfFile}; @@ -69,12 +69,19 @@ enum StdSource { } fn src<T>(fd: libc::c_int, readable: bool, f: |StdSource| -> T) -> T { - with_local_io(|io| { - match io.tty_open(fd, readable) { - Ok(tty) => Some(f(TTY(tty))), - Err(_) => Some(f(File(io.fs_from_raw_fd(fd, DontClose)))), + let mut io = LocalIo::borrow(); + match io.get().tty_open(fd, readable) { + Ok(tty) => f(TTY(tty)), + Err(_) => { + // It's not really that desirable if these handles are closed + // synchronously, and because they're squirreled away in a task + // structure the destructors will be run when the task is + // attempted to get destroyed. This means that if we run a + // synchronous destructor we'll attempt to do some scheduling + // operations which will just result in sadness. + f(File(io.get().fs_from_raw_fd(fd, DontClose))) } - }).unwrap() + } } /// Creates a new non-blocking handle to the stdin of the current process. diff --git a/src/libstd/io/timer.rs b/src/libstd/io/timer.rs index 8dda7935888..202e02d55d0 100644 --- a/src/libstd/io/timer.rs +++ b/src/libstd/io/timer.rs @@ -42,7 +42,7 @@ use comm::{Port, PortOne}; use option::{Option, Some, None}; use result::{Ok, Err}; use io::io_error; -use rt::rtio::{IoFactory, RtioTimer, with_local_io}; +use rt::rtio::{IoFactory, LocalIo, RtioTimer}; pub struct Timer { priv obj: ~RtioTimer @@ -60,17 +60,15 @@ impl Timer { /// for a number of milliseconds, or to possibly create channels which will /// get notified after an amount of time has passed. pub fn new() -> Option<Timer> { - with_local_io(|io| { - match io.timer_init() { - Ok(t) => Some(Timer { obj: t }), - Err(ioerr) => { - debug!("Timer::init: failed to init: {:?}", ioerr); - io_error::cond.raise(ioerr); - None - } + let mut io = LocalIo::borrow(); + match io.get().timer_init() { + Ok(t) => Some(Timer { obj: t }), + Err(ioerr) => { + debug!("Timer::init: failed to init: {:?}", ioerr); + io_error::cond.raise(ioerr); + None } - - }) + } } /// Blocks the current task for `msecs` milliseconds. diff --git a/src/libstd/os.rs b/src/libstd/os.rs index ff939310865..bcd353bab7a 100644 --- a/src/libstd/os.rs +++ b/src/libstd/os.rs @@ -1477,7 +1477,9 @@ mod tests { assert!(*chunk.data == 0xbe); close(fd); } - io::ignore_io_error(|| fs::unlink(&path)); + + let _guard = io::ignore_io_error(); + fs::unlink(&path); } // More recursive_mkdir tests are in extra::tempfile diff --git a/src/libstd/rt/basic.rs b/src/libstd/rt/basic.rs index 311138d15a2..fa47ceb1c04 100644 --- a/src/libstd/rt/basic.rs +++ b/src/libstd/rt/basic.rs @@ -159,8 +159,9 @@ impl EventLoop for BasicLoop { ~BasicRemote::new(self.messages.clone(), id) as ~RemoteCallback } - fn io<'a>(&'a mut self, f: |&'a mut IoFactory|) { - f(self.io) + fn io<'a>(&'a mut self) -> Option<&'a mut IoFactory> { + let factory: &mut IoFactory = self.io; + Some(factory) } } diff --git a/src/libstd/rt/borrowck.rs b/src/libstd/rt/borrowck.rs index 30c2264bd86..82f92bdb803 100644 --- a/src/libstd/rt/borrowck.rs +++ b/src/libstd/rt/borrowck.rs @@ -8,7 +8,6 @@ // option. This file may not be copied, modified, or distributed // except according to those terms. -use cell::Cell; use c_str::{ToCStr, CString}; use libc::{c_char, size_t}; use option::{Option, None, Some}; @@ -35,7 +34,8 @@ pub struct BorrowRecord { } fn try_take_task_borrow_list() -> Option<~[BorrowRecord]> { - Local::borrow(|task: &mut Task| task.borrow_list.take()) + let mut task = Local::borrow(None::<Task>); + task.get().borrow_list.take() } fn swap_task_borrow_list(f: |~[BorrowRecord]| -> ~[BorrowRecord]) { @@ -44,8 +44,9 @@ fn swap_task_borrow_list(f: |~[BorrowRecord]| -> ~[BorrowRecord]) { None => ~[] }; let borrows = f(borrows); - let borrows = Cell::new(borrows); - Local::borrow(|task: &mut Task| task.borrow_list = Some(borrows.take())) + + let mut task = Local::borrow(None::<Task>); + task.get().borrow_list = Some(borrows) } pub fn clear_task_borrow_list() { diff --git a/src/libstd/rt/comm.rs b/src/libstd/rt/comm.rs index 06743bce9bf..2b1e7865a73 100644 --- a/src/libstd/rt/comm.rs +++ b/src/libstd/rt/comm.rs @@ -25,7 +25,7 @@ use unstable::sync::UnsafeArc; use util; use util::Void; use comm::{GenericChan, GenericSmartChan, GenericPort, Peekable, SendDeferred}; -use cell::{Cell, RefCell}; +use cell::RefCell; use clone::Clone; use tuple::ImmutableTuple; @@ -169,10 +169,8 @@ impl<T: Send> ChanOne<T> { Scheduler::run_task(woken_task); }); } else { - let recvr = Cell::new(recvr); - Local::borrow(|sched: &mut Scheduler| { - sched.enqueue_blocked_task(recvr.take()); - }) + let mut sched = Local::borrow(None::<Scheduler>); + sched.get().enqueue_blocked_task(recvr); } } } @@ -230,9 +228,8 @@ impl<T: Send> SelectInner for PortOne<T> { // The optimistic check is never necessary for correctness. For testing // purposes, making it randomly return false simulates a racing sender. use rand::{Rand}; - let actually_check = Local::borrow(|sched: &mut Scheduler| { - Rand::rand(&mut sched.rng) - }); + let mut sched = Local::borrow(None::<Scheduler>); + let actually_check = Rand::rand(&mut sched.get().rng); if actually_check { unsafe { (*self.packet()).state.load(Acquire) == STATE_ONE } } else { @@ -568,7 +565,7 @@ impl<'self, T: Send> SelectPortInner<T> for &'self Port<T> { impl<'self, T: Send> SelectPort<T> for &'self Port<T> { } pub struct SharedChan<T> { - // Just like Chan, but a shared AtomicOption instead of Cell + // Just like Chan, but a shared AtomicOption priv next: UnsafeArc<AtomicOption<StreamChanOne<T>>> } @@ -719,7 +716,6 @@ mod test { use super::*; use option::*; use rt::test::*; - use cell::Cell; use num::Times; use rt::util; @@ -843,9 +839,8 @@ mod test { fn oneshot_multi_task_recv_then_send() { do run_in_newsched_task { let (port, chan) = oneshot::<~int>(); - let port_cell = Cell::new(port); do spawntask { - assert!(port_cell.take().recv() == ~10); + assert!(port.recv() == ~10); } chan.send(~10); @@ -856,13 +851,11 @@ mod test { fn oneshot_multi_task_recv_then_close() { do run_in_newsched_task { let (port, chan) = oneshot::<~int>(); - let port_cell = Cell::new(port); - let chan_cell = Cell::new(chan); do spawntask_later { - let _cell = chan_cell.take(); + let _ = chan; } let res = do spawntask_try { - assert!(port_cell.take().recv() == ~10); + assert!(port.recv() == ~10); }; assert!(res.is_err()); } @@ -874,9 +867,8 @@ mod test { stress_factor().times(|| { do run_in_newsched_task { let (port, chan) = oneshot::<int>(); - let port_cell = Cell::new(port); let thread = do spawntask_thread { - let _p = port_cell.take(); + let _ = port; }; let _chan = chan; thread.join(); @@ -890,14 +882,11 @@ mod test { stress_factor().times(|| { do run_in_newsched_task { let (port, chan) = oneshot::<int>(); - let chan_cell = Cell::new(chan); - let port_cell = Cell::new(port); let thread1 = do spawntask_thread { - let _p = port_cell.take(); + let _ = port; }; let thread2 = do spawntask_thread { - let c = chan_cell.take(); - c.send(1); + chan.send(1); }; thread1.join(); thread2.join(); @@ -911,19 +900,17 @@ mod test { stress_factor().times(|| { do run_in_newsched_task { let (port, chan) = oneshot::<int>(); - let chan_cell = Cell::new(chan); - let port_cell = Cell::new(port); let thread1 = do spawntask_thread { - let port_cell = Cell::new(port_cell.take()); + let port = port; let res = do spawntask_try { - port_cell.take().recv(); + port.recv(); }; assert!(res.is_err()); }; let thread2 = do spawntask_thread { - let chan_cell = Cell::new(chan_cell.take()); + let chan = chan; do spawntask { - chan_cell.take(); + let _ = chan; } }; thread1.join(); @@ -938,13 +925,11 @@ mod test { stress_factor().times(|| { do run_in_newsched_task { let (port, chan) = oneshot::<~int>(); - let chan_cell = Cell::new(chan); - let port_cell = Cell::new(port); let thread1 = do spawntask_thread { - chan_cell.take().send(~10); + chan.send(~10); }; let thread2 = do spawntask_thread { - assert!(port_cell.take().recv() == ~10); + assert!(port.recv() == ~10); }; thread1.join(); thread2.join(); @@ -965,9 +950,7 @@ mod test { fn send(chan: Chan<~int>, i: int) { if i == 10 { return } - let chan_cell = Cell::new(chan); do spawntask_random { - let chan = chan_cell.take(); chan.send(~i); send(chan, i + 1); } @@ -976,9 +959,7 @@ mod test { fn recv(port: Port<~int>, i: int) { if i == 10 { return } - let port_cell = Cell::new(port); do spawntask_random { - let port = port_cell.take(); assert!(port.recv() == ~i); recv(port, i + 1); }; @@ -1131,7 +1112,7 @@ mod test { #[test] fn send_deferred() { - use unstable::sync::atomically; + use unstable::sync::atomic; // Tests no-rescheduling of send_deferred on all types of channels. do run_in_newsched_task { @@ -1141,24 +1122,18 @@ mod test { let cshared = SharedChan::new(cshared); let mp = megapipe(); - let pone = Cell::new(pone); - do spawntask { pone.take().recv(); } - let pstream = Cell::new(pstream); - do spawntask { pstream.take().recv(); } - let pshared = Cell::new(pshared); - do spawntask { pshared.take().recv(); } - let p_mp = Cell::new(mp.clone()); - do spawntask { p_mp.take().recv(); } + do spawntask { pone.recv(); } + do spawntask { pstream.recv(); } + do spawntask { pshared.recv(); } + let p_mp = mp.clone(); + do spawntask { p_mp.recv(); } - let cs = Cell::new((cone, cstream, cshared, mp)); unsafe { - atomically(|| { - let (cone, cstream, cshared, mp) = cs.take(); - cone.send_deferred(()); - cstream.send_deferred(()); - cshared.send_deferred(()); - mp.send_deferred(()); - }) + let _guard = atomic(); + cone.send_deferred(()); + cstream.send_deferred(()); + cshared.send_deferred(()); + mp.send_deferred(()); } } } diff --git a/src/libstd/rt/deque.rs b/src/libstd/rt/deque.rs index 3595587f342..44672984b64 100644 --- a/src/libstd/rt/deque.rs +++ b/src/libstd/rt/deque.rs @@ -163,10 +163,9 @@ impl<T: Send> BufferPool<T> { fn free(&mut self, buf: ~Buffer<T>) { unsafe { - use cell::Cell; - let buf = Cell::new(buf); + let mut buf = Some(buf); self.pool.with(|pool| { - let buf = buf.take(); + let buf = buf.take_unwrap(); match pool.iter().position(|v| v.size() > buf.size()) { Some(i) => pool.insert(i, buf), None => pool.push(buf), diff --git a/src/libstd/rt/kill.rs b/src/libstd/rt/kill.rs index 56c77ffaa0d..e3f9cd09632 100644 --- a/src/libstd/rt/kill.rs +++ b/src/libstd/rt/kill.rs @@ -151,7 +151,6 @@ There are two known issues with the current scheme for exit code propagation. */ use cast; -use cell::Cell; use option::{Option, Some, None}; use prelude::*; use rt::task::Task; @@ -256,8 +255,10 @@ impl Death { /// Collect failure exit codes from children and propagate them to a parent. pub fn collect_failure(&mut self, result: UnwindResult) { - let result = Cell::new(result); - self.on_exit.take().map(|on_exit| on_exit(result.take())); + match self.on_exit.take() { + None => {} + Some(on_exit) => on_exit(result), + } } /// Enter a possibly-nested "atomic" section of code. Just for assertions. diff --git a/src/libstd/rt/local.rs b/src/libstd/rt/local.rs index 2375ce55766..d73ad98a25b 100644 --- a/src/libstd/rt/local.rs +++ b/src/libstd/rt/local.rs @@ -12,36 +12,28 @@ use option::{Option, Some, None}; use rt::sched::Scheduler; use rt::task::Task; use rt::local_ptr; -use cell::Cell; -pub trait Local { +/// Encapsulates some task-local data. +pub trait Local<Borrowed> { fn put(value: ~Self); fn take() -> ~Self; fn exists(unused_value: Option<Self>) -> bool; - fn borrow<T>(f: |&mut Self| -> T) -> T; + fn borrow(unused_value: Option<Self>) -> Borrowed; unsafe fn unsafe_take() -> ~Self; unsafe fn unsafe_borrow() -> *mut Self; unsafe fn try_unsafe_borrow() -> Option<*mut Self>; } -impl Local for Task { +impl Local<local_ptr::Borrowed<Task>> for Task { #[inline] fn put(value: ~Task) { unsafe { local_ptr::put(value) } } #[inline] fn take() -> ~Task { unsafe { local_ptr::take() } } fn exists(_: Option<Task>) -> bool { local_ptr::exists() } - fn borrow<T>(f: |&mut Task| -> T) -> T { - let mut res: Option<T> = None; - let res_ptr: *mut Option<T> = &mut res; + #[inline] + fn borrow(_: Option<Task>) -> local_ptr::Borrowed<Task> { unsafe { - local_ptr::borrow(|task| { - let result = f(task); - *res_ptr = Some(result); - }) - } - match res { - Some(r) => { r } - None => { rtabort!("function failed in local_borrow") } + local_ptr::borrow::<Task>() } } #[inline] @@ -54,13 +46,35 @@ impl Local for Task { } } -impl Local for Scheduler { +/// Encapsulates a temporarily-borrowed scheduler. +pub struct BorrowedScheduler { + priv task: local_ptr::Borrowed<Task>, +} + +impl BorrowedScheduler { + fn new(mut task: local_ptr::Borrowed<Task>) -> BorrowedScheduler { + if task.get().sched.is_none() { + rtabort!("no scheduler") + } else { + BorrowedScheduler { + task: task, + } + } + } + + #[inline] + pub fn get<'a>(&'a mut self) -> &'a mut ~Scheduler { + match self.task.get().sched { + None => rtabort!("no scheduler"), + Some(ref mut sched) => sched, + } + } +} + +impl Local<BorrowedScheduler> for Scheduler { fn put(value: ~Scheduler) { - let value = Cell::new(value); - Local::borrow(|task: &mut Task| { - let task = task; - task.sched = Some(value.take()); - }); + let mut task = Local::borrow(None::<Task>); + task.get().sched = Some(value); } #[inline] fn take() -> ~Scheduler { @@ -71,24 +85,12 @@ impl Local for Scheduler { } } fn exists(_: Option<Scheduler>) -> bool { - Local::borrow(|task: &mut Task| { - match task.sched { - Some(ref _task) => true, - None => false - } - }) + let mut task = Local::borrow(None::<Task>); + task.get().sched.is_some() } - fn borrow<T>(f: |&mut Scheduler| -> T) -> T { - Local::borrow(|task: &mut Task| { - match task.sched { - Some(~ref mut task) => { - f(task) - } - None => { - rtabort!("no scheduler") - } - } - }) + #[inline] + fn borrow(_: Option<Scheduler>) -> BorrowedScheduler { + BorrowedScheduler::new(Local::borrow(None::<Task>)) } unsafe fn unsafe_take() -> ~Scheduler { rtabort!("unimpl") } unsafe fn unsafe_borrow() -> *mut Scheduler { @@ -182,11 +184,11 @@ mod test { let task = ~Task::new_root(&mut sched.stack_pool, None, proc(){}); Local::put(task); - let res = Local::borrow(|_task: &mut Task| { - true - }); - assert!(res) - let task: ~Task = Local::take(); + { + let _ = Local::borrow(None::<Task>); + } + + let task: ~Task = Local::take(); cleanup_task(task); } } diff --git a/src/libstd/rt/local_heap.rs b/src/libstd/rt/local_heap.rs index 2386a261bdf..e364137de45 100644 --- a/src/libstd/rt/local_heap.rs +++ b/src/libstd/rt/local_heap.rs @@ -304,7 +304,8 @@ pub unsafe fn local_free(ptr: *libc::c_char) { } pub fn live_allocs() -> *mut Box { - Local::borrow(|task: &mut Task| task.heap.live_allocs) + let mut task = Local::borrow(None::<Task>); + task.get().heap.live_allocs } #[cfg(test)] diff --git a/src/libstd/rt/local_ptr.rs b/src/libstd/rt/local_ptr.rs index be3b5f951eb..66fe9742121 100644 --- a/src/libstd/rt/local_ptr.rs +++ b/src/libstd/rt/local_ptr.rs @@ -18,8 +18,7 @@ #[allow(dead_code)]; use cast; -use cell::Cell; -use unstable::finally::Finally; +use ops::Drop; #[cfg(windows)] // mingw-w32 doesn't like thread_local things #[cfg(target_os = "android")] // see #10686 @@ -28,20 +27,48 @@ pub use self::native::*; #[cfg(not(windows), not(target_os = "android"))] pub use self::compiled::*; +/// Encapsulates a borrowed value. When this value goes out of scope, the +/// pointer is returned. +pub struct Borrowed<T> { + priv val: *(), +} + +#[unsafe_destructor] +impl<T> Drop for Borrowed<T> { + fn drop(&mut self) { + unsafe { + if self.val.is_null() { + rtabort!("Aiee, returning null borrowed object!"); + } + let val: ~T = cast::transmute(self.val); + put::<T>(val); + assert!(exists()); + } + } +} + +impl<T> Borrowed<T> { + pub fn get<'a>(&'a mut self) -> &'a mut T { + unsafe { + let val_ptr: &mut ~T = cast::transmute(&mut self.val); + let val_ptr: &'a mut T = *val_ptr; + val_ptr + } + } +} + /// Borrow the thread-local value from thread-local storage. /// While the value is borrowed it is not available in TLS. /// /// # Safety note /// /// Does not validate the pointer type. -pub unsafe fn borrow<T>(f: |&mut T|) { - let mut value = take(); - - // XXX: Need a different abstraction from 'finally' here to avoid unsafety - let unsafe_ptr = cast::transmute_mut_region(&mut *value); - let value_cell = Cell::new(value); - - (|| f(unsafe_ptr)).finally(|| put(value_cell.take())); +#[inline] +pub unsafe fn borrow<T>() -> Borrowed<T> { + let val: *() = cast::transmute(take::<T>()); + Borrowed { + val: val, + } } /// Compiled implementation of accessing the runtime local pointer. This is diff --git a/src/libstd/rt/mod.rs b/src/libstd/rt/mod.rs index 3d0222cefae..ce8d1ab1983 100644 --- a/src/libstd/rt/mod.rs +++ b/src/libstd/rt/mod.rs @@ -57,7 +57,6 @@ Several modules in `core` are clients of `rt`: // XXX: this should not be here. #[allow(missing_doc)]; -use cell::Cell; use clone::Clone; use container::Container; use iter::Iterator; @@ -274,7 +273,7 @@ fn run_(main: proc(), use_main_sched: bool) -> int { let nscheds = util::default_sched_threads(); - let main = Cell::new(main); + let mut main = Some(main); // The shared list of sleeping schedulers. let sleepers = SleeperList::new(); @@ -376,24 +375,24 @@ fn run_(main: proc(), use_main_sched: bool) -> int { }; let mut threads = ~[]; - - let on_exit = Cell::new(on_exit); + let mut on_exit = Some(on_exit); if !use_main_sched { // In the case where we do not use a main_thread scheduler we // run the main task in one of our threads. - let mut main_task = ~Task::new_root(&mut scheds[0].stack_pool, None, main.take()); + let mut main_task = ~Task::new_root(&mut scheds[0].stack_pool, + None, + ::util::replace(&mut main, + None).unwrap()); main_task.name = Some(SendStrStatic("<main>")); - main_task.death.on_exit = Some(on_exit.take()); - let main_task_cell = Cell::new(main_task); + main_task.death.on_exit = ::util::replace(&mut on_exit, None); let sched = scheds.pop(); - let sched_cell = Cell::new(sched); + let main_task = main_task; let thread = do Thread::start { - let sched = sched_cell.take(); - sched.bootstrap(main_task_cell.take()); + sched.bootstrap(main_task); }; threads.push(thread); } @@ -401,9 +400,8 @@ fn run_(main: proc(), use_main_sched: bool) -> int { // Run each remaining scheduler in a thread. for sched in scheds.move_rev_iter() { rtdebug!("creating regular schedulers"); - let sched_cell = Cell::new(sched); let thread = do Thread::start { - let mut sched = sched_cell.take(); + let mut sched = sched; let bootstrap_task = ~do Task::new_root(&mut sched.stack_pool, None) || { rtdebug!("boostraping a non-primary scheduler"); }; @@ -415,16 +413,19 @@ fn run_(main: proc(), use_main_sched: bool) -> int { // If we do have a main thread scheduler, run it now. if use_main_sched { - rtdebug!("about to create the main scheduler task"); let mut main_sched = main_sched.unwrap(); let home = Sched(main_sched.make_handle()); - let mut main_task = ~Task::new_root_homed(&mut main_sched.stack_pool, None, - home, main.take()); + let mut main_task = ~Task::new_root_homed(&mut main_sched.stack_pool, + None, + home, + ::util::replace(&mut main, + None). + unwrap()); main_task.name = Some(SendStrStatic("<main>")); - main_task.death.on_exit = Some(on_exit.take()); + main_task.death.on_exit = ::util::replace(&mut on_exit, None); rtdebug!("bootstrapping main_task"); main_sched.bootstrap(main_task); diff --git a/src/libstd/rt/rtio.rs b/src/libstd/rt/rtio.rs index f83932f9ffa..05cc051a23e 100644 --- a/src/libstd/rt/rtio.rs +++ b/src/libstd/rt/rtio.rs @@ -9,15 +9,19 @@ // except according to those terms. use c_str::CString; +use cast; use comm::{SharedChan, PortOne, Port}; use libc::c_int; use libc; +use ops::Drop; use option::*; use path::Path; use result::*; use ai = io::net::addrinfo; use io::IoError; +use io::native::NATIVE_IO_FACTORY; +use io::native; use io::net::ip::{IpAddr, SocketAddr}; use io::process::{ProcessConfig, ProcessExit}; use io::signal::Signum; @@ -34,9 +38,8 @@ pub trait EventLoop { fn pausible_idle_callback(&mut self, ~Callback) -> ~PausibleIdleCallback; fn remote_callback(&mut self, ~Callback) -> ~RemoteCallback; - /// The asynchronous I/O services. Not all event loops may provide one - // FIXME(#9382) this is an awful interface - fn io<'a>(&'a mut self, f: |&'a mut IoFactory|); + /// The asynchronous I/O services. Not all event loops may provide one. + fn io<'a>(&'a mut self) -> Option<&'a mut IoFactory>; } pub trait RemoteCallback { @@ -75,31 +78,60 @@ pub enum CloseBehavior { CloseAsynchronously, } -pub fn with_local_io<T>(f: |&mut IoFactory| -> Option<T>) -> Option<T> { - use rt::sched::Scheduler; - use rt::local::Local; - use io::native; - - unsafe { - // First, attempt to use the local scheduler's I/O services - let sched: Option<*mut Scheduler> = Local::try_unsafe_borrow(); - match sched { - Some(sched) => { - let mut io = None; - (*sched).event_loop.io(|i| io = Some(i)); - match io { - Some(io) => return f(io), - None => {} +pub struct LocalIo<'a> { + priv factory: &'a mut IoFactory, +} + +#[unsafe_destructor] +impl<'a> Drop for LocalIo<'a> { + fn drop(&mut self) { + // XXX(pcwalton): Do nothing here for now, but eventually we may want + // something. For now this serves to make `LocalIo` noncopyable. + } +} + +impl<'a> LocalIo<'a> { + /// Returns the local I/O: either the local scheduler's I/O services or + /// the native I/O services. + pub fn borrow() -> LocalIo { + use rt::sched::Scheduler; + use rt::local::Local; + + unsafe { + // First, attempt to use the local scheduler's I/O services + let sched: Option<*mut Scheduler> = Local::try_unsafe_borrow(); + match sched { + Some(sched) => { + match (*sched).event_loop.io() { + Some(factory) => { + return LocalIo { + factory: factory, + } + } + None => {} + } } + None => {} + } + // If we don't have a scheduler or the scheduler doesn't have I/O + // services, then fall back to the native I/O services. + let native_io: &'static mut native::IoFactory = + &mut NATIVE_IO_FACTORY; + LocalIo { + factory: native_io as &mut IoFactory:'static } - None => {} } } - // If we don't have a scheduler or the scheduler doesn't have I/O services, - // then fall back to the native I/O services. - let mut io = native::IoFactory; - f(&mut io as &mut IoFactory) + /// Returns the underlying I/O factory as a trait reference. + #[inline] + pub fn get<'a>(&'a mut self) -> &'a mut IoFactory { + // XXX(pcwalton): I think this is actually sound? Could borrow check + // allow this safely? + unsafe { + cast::transmute_copy(&self.factory) + } + } } pub trait IoFactory { diff --git a/src/libstd/rt/sched.rs b/src/libstd/rt/sched.rs index eb5b2df4ed9..eefed412466 100644 --- a/src/libstd/rt/sched.rs +++ b/src/libstd/rt/sched.rs @@ -24,7 +24,6 @@ use rt::local_ptr; use rt::local::Local; use rt::rtio::{RemoteCallback, PausibleIdleCallback, Callback}; use borrow::{to_uint}; -use cell::Cell; use rand::{XorShiftRng, Rng, Rand}; use iter::range; use unstable::mutex::Mutex; @@ -235,12 +234,12 @@ impl Scheduler { unsafe { let event_loop: *mut ~EventLoop = &mut self.event_loop; - // Our scheduler must be in the task before the event loop - // is started. - let self_sched = Cell::new(self); - Local::borrow(|stask: &mut Task| { - stask.sched = Some(self_sched.take()); - }); + { + // Our scheduler must be in the task before the event loop + // is started. + let mut stask = Local::borrow(None::<Task>); + stask.get().sched = Some(self); + } (*event_loop).run(); } @@ -751,10 +750,8 @@ impl Scheduler { } pub fn run_task_later(next_task: ~Task) { - let next_task = Cell::new(next_task); - Local::borrow(|sched: &mut Scheduler| { - sched.enqueue_task(next_task.take()); - }); + let mut sched = Local::borrow(None::<Scheduler>); + sched.get().enqueue_task(next_task); } /// Yield control to the scheduler, executing another task. This is guaranteed @@ -922,7 +919,6 @@ mod test { use unstable::run_in_bare_thread; use borrow::to_uint; use rt::sched::{Scheduler}; - use cell::Cell; use rt::deque::BufferPool; use rt::thread::Thread; use rt::task::{Task, Sched}; @@ -1048,7 +1044,7 @@ mod test { queues.clone(), sleepers.clone()); - let normal_handle = Cell::new(normal_sched.make_handle()); + let normal_handle = normal_sched.make_handle(); let friend_handle = normal_sched.make_handle(); @@ -1061,7 +1057,7 @@ mod test { false, Some(friend_handle)); - let special_handle = Cell::new(special_sched.make_handle()); + let special_handle = special_sched.make_handle(); let t1_handle = special_sched.make_handle(); let t4_handle = special_sched.make_handle(); @@ -1092,26 +1088,19 @@ mod test { }; rtdebug!("task4 id: **{}**", borrow::to_uint(task4)); - let task1 = Cell::new(task1); - let task2 = Cell::new(task2); - let task3 = Cell::new(task3); - let task4 = Cell::new(task4); - // Signal from the special task that we are done. let (port, chan) = oneshot::<()>(); - let port = Cell::new(port); - let chan = Cell::new(chan); let normal_task = ~do Task::new_root(&mut normal_sched.stack_pool, None) { rtdebug!("*about to submit task2*"); - Scheduler::run_task(task2.take()); + Scheduler::run_task(task2); rtdebug!("*about to submit task4*"); - Scheduler::run_task(task4.take()); + Scheduler::run_task(task4); rtdebug!("*normal_task done*"); - port.take().recv(); - let mut nh = normal_handle.take(); + port.recv(); + let mut nh = normal_handle; nh.send(Shutdown); - let mut sh = special_handle.take(); + let mut sh = special_handle; sh.send(Shutdown); }; @@ -1119,27 +1108,24 @@ mod test { let special_task = ~do Task::new_root(&mut special_sched.stack_pool, None) { rtdebug!("*about to submit task1*"); - Scheduler::run_task(task1.take()); + Scheduler::run_task(task1); rtdebug!("*about to submit task3*"); - Scheduler::run_task(task3.take()); + Scheduler::run_task(task3); rtdebug!("*done with special_task*"); - chan.take().send(()); + chan.send(()); }; rtdebug!("special task: {}", borrow::to_uint(special_task)); - let special_sched = Cell::new(special_sched); - let normal_sched = Cell::new(normal_sched); - let special_task = Cell::new(special_task); - let normal_task = Cell::new(normal_task); - + let normal_sched = normal_sched; let normal_thread = do Thread::start { - normal_sched.take().bootstrap(normal_task.take()); + normal_sched.bootstrap(normal_task); rtdebug!("finished with normal_thread"); }; + let special_sched = special_sched; let special_thread = do Thread::start { - special_sched.take().bootstrap(special_task.take()); + special_sched.bootstrap(special_task); rtdebug!("finished with special_sched"); }; @@ -1178,20 +1164,18 @@ mod test { do run_in_bare_thread { let (port, chan) = oneshot::<()>(); - let port = Cell::new(port); - let chan = Cell::new(chan); let thread_one = do Thread::start { - let chan = Cell::new(chan.take()); + let chan = chan; do run_in_newsched_task_core { - chan.take().send(()); + chan.send(()); } }; let thread_two = do Thread::start { - let port = Cell::new(port.take()); + let port = port; do run_in_newsched_task_core { - port.take().recv(); + port.recv(); } }; @@ -1222,10 +1206,9 @@ mod test { let mut handle = sched.make_handle(); - let sched = Cell::new(sched); - + let sched = sched; let thread = do Thread::start { - let mut sched = sched.take(); + let mut sched = sched; let bootstrap_task = ~Task::new_root(&mut sched.stack_pool, None, @@ -1256,9 +1239,8 @@ mod test { let mut ports = ~[]; 10.times(|| { let (port, chan) = oneshot(); - let chan_cell = Cell::new(chan); do spawntask_later { - chan_cell.take().send(()); + chan.send(()); } ports.push(port); }); diff --git a/src/libstd/rt/task.rs b/src/libstd/rt/task.rs index 63cc397e8d7..86cc895eb27 100644 --- a/src/libstd/rt/task.rs +++ b/src/libstd/rt/task.rs @@ -19,7 +19,6 @@ use prelude::*; use borrow; use cast::transmute; -use cell::Cell; use cleanup; use libc::{c_void, uintptr_t, c_char, size_t}; use local_data; @@ -144,17 +143,15 @@ impl Task { f: proc(), home: SchedHome) -> ~Task { - let f = Cell::new(f); - let home = Cell::new(home); - Local::borrow(|running_task: &mut Task| { - let mut sched = running_task.sched.take_unwrap(); - let new_task = ~running_task.new_child_homed(&mut sched.stack_pool, - stack_size, - home.take(), - f.take()); - running_task.sched = Some(sched); - new_task - }) + let mut running_task = Local::borrow(None::<Task>); + let mut sched = running_task.get().sched.take_unwrap(); + let new_task = ~running_task.get() + .new_child_homed(&mut sched.stack_pool, + stack_size, + home, + f); + running_task.get().sched = Some(sched); + new_task } pub fn build_child(stack_size: Option<uint>, f: proc()) -> ~Task { @@ -165,17 +162,14 @@ impl Task { f: proc(), home: SchedHome) -> ~Task { - let f = Cell::new(f); - let home = Cell::new(home); - Local::borrow(|running_task: &mut Task| { - let mut sched = running_task.sched.take_unwrap(); - let new_task = ~Task::new_root_homed(&mut sched.stack_pool, - stack_size, - home.take(), - f.take()); - running_task.sched = Some(sched); - new_task - }) + let mut running_task = Local::borrow(None::<Task>); + let mut sched = running_task.get().sched.take_unwrap(); + let new_task = ~Task::new_root_homed(&mut sched.stack_pool, + stack_size, + home, + f); + running_task.get().sched = Some(sched); + new_task } pub fn build_root(stack_size: Option<uint>, f: proc()) -> ~Task { @@ -371,26 +365,25 @@ impl Task { // Grab both the scheduler and the task from TLS and check if the // task is executing on an appropriate scheduler. pub fn on_appropriate_sched() -> bool { - Local::borrow(|task: &mut Task| { - let sched_id = task.sched.get_ref().sched_id(); - let sched_run_anything = task.sched.get_ref().run_anything; - match task.task_type { - GreenTask(Some(AnySched)) => { - rtdebug!("anysched task in sched check ****"); - sched_run_anything - } - GreenTask(Some(Sched(SchedHandle { sched_id: ref id, ..}))) => { - rtdebug!("homed task in sched check ****"); - *id == sched_id - } - GreenTask(None) => { - rtabort!("task without home"); - } - SchedTask => { - rtabort!("type error: expected: GreenTask, found: SchedTask"); - } + let mut task = Local::borrow(None::<Task>); + let sched_id = task.get().sched.get_ref().sched_id(); + let sched_run_anything = task.get().sched.get_ref().run_anything; + match task.get().task_type { + GreenTask(Some(AnySched)) => { + rtdebug!("anysched task in sched check ****"); + sched_run_anything + } + GreenTask(Some(Sched(SchedHandle { sched_id: ref id, ..}))) => { + rtdebug!("homed task in sched check ****"); + *id == sched_id + } + GreenTask(None) => { + rtabort!("task without home"); } - }) + SchedTask => { + rtabort!("type error: expected: GreenTask, found: SchedTask"); + } + } } } @@ -433,16 +426,16 @@ impl Coroutine { } fn build_start_wrapper(start: proc()) -> proc() { - let start_cell = Cell::new(start); let wrapper: proc() = proc() { // First code after swap to this new context. Run our // cleanup job. unsafe { // Again - might work while safe, or it might not. - Local::borrow(|sched: &mut Scheduler| { - sched.run_cleanup_job(); - }); + { + let mut sched = Local::borrow(None::<Scheduler>); + sched.get().run_cleanup_job(); + } // To call the run method on a task we need a direct // reference to it. The task is in TLS, so we can @@ -451,6 +444,7 @@ impl Coroutine { // need to unsafe_borrow. let task: *mut Task = Local::unsafe_borrow(); + let mut start_cell = Some(start); (*task).run(|| { // N.B. Removing `start` from the start wrapper // closure by emptying a cell is critical for @@ -462,7 +456,7 @@ impl Coroutine { // be in task context. By moving `start` out of // the closure, all the user code goes our of // scope while the task is still running. - let start = start_cell.take(); + let start = start_cell.take_unwrap(); start(); }); } @@ -594,16 +588,19 @@ pub extern "C" fn rust_stack_exhausted() { // #2361 - possible implementation of not using landing pads if in_green_task_context() { - Local::borrow(|task: &mut Task| { - let n = task.name.as_ref().map(|n| n.as_slice()).unwrap_or("<unnamed>"); - - // See the message below for why this is not emitted to the - // task's logger. This has the additional conundrum of the - // logger may not be initialized just yet, meaning that an FFI - // call would happen to initialized it (calling out to libuv), - // and the FFI call needs 2MB of stack when we just ran out. - rterrln!("task '{}' has overflowed its stack", n); - }) + let mut task = Local::borrow(None::<Task>); + let n = task.get() + .name + .as_ref() + .map(|n| n.as_slice()) + .unwrap_or("<unnamed>"); + + // See the message below for why this is not emitted to the + // task's logger. This has the additional conundrum of the + // logger may not be initialized just yet, meaning that an FFI + // call would happen to initialized it (calling out to libuv), + // and the FFI call needs 2MB of stack when we just ran out. + rterrln!("task '{}' has overflowed its stack", n); } else { rterrln!("stack overflow in non-task context"); } diff --git a/src/libstd/rt/test.rs b/src/libstd/rt/test.rs index 2ab543a4c36..96b80d11129 100644 --- a/src/libstd/rt/test.rs +++ b/src/libstd/rt/test.rs @@ -10,7 +10,6 @@ use io::net::ip::{SocketAddr, Ipv4Addr, Ipv6Addr}; -use cell::Cell; use clone::Clone; use container::Container; use iter::{Iterator, range}; @@ -65,16 +64,14 @@ pub fn new_test_sched() -> Scheduler { } pub fn run_in_uv_task(f: proc()) { - let f = Cell::new(f); do run_in_bare_thread { - run_in_uv_task_core(f.take()); + run_in_uv_task_core(f); } } pub fn run_in_newsched_task(f: proc()) { - let f = Cell::new(f); do run_in_bare_thread { - run_in_newsched_task_core(f.take()); + run_in_newsched_task_core(f); } } @@ -206,8 +203,6 @@ pub fn run_in_mt_newsched_task(f: proc()) { // see comment in other function (raising fd limits) prepare_for_lots_of_tests(); - let f = Cell::new(f); - do run_in_bare_thread { let nthreads = match os::getenv("RUST_RT_TEST_THREADS") { Some(nstr) => FromStr::from_str(nstr).unwrap(), @@ -254,18 +249,18 @@ pub fn run_in_mt_newsched_task(f: proc()) { rtassert!(exit_status.is_success()); }; - let mut main_task = ~Task::new_root(&mut scheds[0].stack_pool, None, f.take()); + let mut main_task = ~Task::new_root(&mut scheds[0].stack_pool, + None, + f); main_task.death.on_exit = Some(on_exit); let mut threads = ~[]; - let main_task = Cell::new(main_task); let main_thread = { let sched = scheds.pop(); - let sched_cell = Cell::new(sched); + let main_task = main_task; do Thread::start { - let sched = sched_cell.take(); - sched.bootstrap(main_task.take()); + sched.bootstrap(main_task); } }; threads.push(main_thread); @@ -275,11 +270,9 @@ pub fn run_in_mt_newsched_task(f: proc()) { let bootstrap_task = ~do Task::new_root(&mut sched.stack_pool, None) || { rtdebug!("bootstrapping non-primary scheduler"); }; - let bootstrap_task_cell = Cell::new(bootstrap_task); - let sched_cell = Cell::new(sched); + let sched = sched; let thread = do Thread::start { - let sched = sched_cell.take(); - sched.bootstrap(bootstrap_task_cell.take()); + sched.bootstrap(bootstrap_task); }; threads.push(thread); @@ -335,11 +328,8 @@ pub fn spawntask_try(f: proc()) -> Result<(),()> { /// Spawn a new task in a new scheduler and return a thread handle. pub fn spawntask_thread(f: proc()) -> Thread<()> { - - let f = Cell::new(f); - let thread = do Thread::start { - run_in_newsched_task_core(f.take()); + run_in_newsched_task_core(f); }; return thread; diff --git a/src/libstd/rt/tube.rs b/src/libstd/rt/tube.rs index 0d4171d5a64..5e867bcdfba 100644 --- a/src/libstd/rt/tube.rs +++ b/src/libstd/rt/tube.rs @@ -88,7 +88,6 @@ impl<T> Clone for Tube<T> { #[cfg(test)] mod test { - use cell::Cell; use rt::test::*; use rt::rtio::EventLoop; use rt::sched::Scheduler; @@ -100,11 +99,10 @@ mod test { fn simple_test() { do run_in_newsched_task { let mut tube: Tube<int> = Tube::new(); - let tube_clone = tube.clone(); - let tube_clone_cell = Cell::new(tube_clone); + let mut tube_clone = Some(tube.clone()); let sched: ~Scheduler = Local::take(); sched.deschedule_running_task_and_then(|sched, task| { - let mut tube_clone = tube_clone_cell.take(); + let mut tube_clone = tube_clone.take_unwrap(); tube_clone.send(1); sched.enqueue_blocked_task(task); }); @@ -117,13 +115,12 @@ mod test { fn blocking_test() { do run_in_newsched_task { let mut tube: Tube<int> = Tube::new(); - let tube_clone = tube.clone(); - let tube_clone = Cell::new(tube_clone); + let mut tube_clone = Some(tube.clone()); let sched: ~Scheduler = Local::take(); sched.deschedule_running_task_and_then(|sched, task| { - let tube_clone = Cell::new(tube_clone.take()); + let tube_clone = tube_clone.take_unwrap(); do sched.event_loop.callback { - let mut tube_clone = tube_clone.take(); + let mut tube_clone = tube_clone; // The task should be blocked on this now and // sending will wake it up. tube_clone.send(1); @@ -141,26 +138,24 @@ mod test { do run_in_newsched_task { let mut tube: Tube<int> = Tube::new(); - let tube_clone = tube.clone(); - let tube_clone = Cell::new(tube_clone); + let mut tube_clone = Some(tube.clone()); let sched: ~Scheduler = Local::take(); sched.deschedule_running_task_and_then(|sched, task| { - callback_send(tube_clone.take(), 0); + callback_send(tube_clone.take_unwrap(), 0); fn callback_send(tube: Tube<int>, i: int) { - if i == 100 { return; } - - let tube = Cell::new(Cell::new(tube)); - Local::borrow(|sched: &mut Scheduler| { - let tube = tube.take(); - do sched.event_loop.callback { - let mut tube = tube.take(); - // The task should be blocked on this now and - // sending will wake it up. - tube.send(i); - callback_send(tube, i + 1); - } - }) + if i == 100 { + return + } + + let mut sched = Local::borrow(None::<Scheduler>); + do sched.get().event_loop.callback { + let mut tube = tube; + // The task should be blocked on this now and + // sending will wake it up. + tube.send(i); + callback_send(tube, i + 1); + } } sched.enqueue_blocked_task(task); diff --git a/src/libstd/run.rs b/src/libstd/run.rs index 2447bba98d6..f4c6cdbd934 100644 --- a/src/libstd/run.rs +++ b/src/libstd/run.rs @@ -12,7 +12,6 @@ #[allow(missing_doc)]; -use cell::Cell; use comm::{stream, SharedChan}; use io::Reader; use io::process::ProcessExit; @@ -212,8 +211,8 @@ impl Process { */ pub fn finish_with_output(&mut self) -> ProcessOutput { self.close_input(); - let output = Cell::new(self.inner.io[1].take()); - let error = Cell::new(self.inner.io[2].take()); + let output = self.inner.io[1].take(); + let error = self.inner.io[2].take(); // Spawn two entire schedulers to read both stdout and sterr // in parallel so we don't deadlock while blocking on one @@ -224,20 +223,20 @@ impl Process { let ch_clone = ch.clone(); do spawn { - io::ignore_io_error(|| { - match error.take() { - Some(ref mut e) => ch.send((2, e.read_to_end())), - None => ch.send((2, ~[])) - } - }) + let _guard = io::ignore_io_error(); + let mut error = error; + match error { + Some(ref mut e) => ch.send((2, e.read_to_end())), + None => ch.send((2, ~[])) + } } do spawn { - io::ignore_io_error(|| { - match output.take() { - Some(ref mut e) => ch_clone.send((1, e.read_to_end())), - None => ch_clone.send((1, ~[])) - } - }) + let _guard = io::ignore_io_error(); + let mut output = output; + match output { + Some(ref mut e) => ch_clone.send((1, e.read_to_end())), + None => ch_clone.send((1, ~[])) + } } let status = self.finish(); diff --git a/src/libstd/select.rs b/src/libstd/select.rs index 43f1c3c5296..01b953051db 100644 --- a/src/libstd/select.rs +++ b/src/libstd/select.rs @@ -10,18 +10,16 @@ #[allow(missing_doc)]; -use cell::Cell; use comm; use container::Container; use iter::{Iterator, DoubleEndedIterator}; +use kinds::Send; +use ops::Drop; use option::*; -// use either::{Either, Left, Right}; -// use rt::kill::BlockedTask; use rt::local::Local; use rt::rtio::EventLoop; use rt::sched::Scheduler; use rt::shouldnt_be_public::{SelectInner, SelectPortInner}; -use unstable::finally::Finally; use vec::{OwnedVector, MutableVector}; /// Trait for message-passing primitives that can be select()ed on. @@ -32,6 +30,18 @@ pub trait Select : SelectInner { } // that implement Select on different types to use select().) pub trait SelectPort<T> : SelectPortInner<T> { } +/// A helper type that throws away a value on a port. +struct PortGuard<T> { + port: Option<comm::PortOne<T>>, +} + +#[unsafe_destructor] +impl<T:Send> Drop for PortGuard<T> { + fn drop(&mut self) { + let _ = self.port.take_unwrap().recv(); + } +} + /// Receive a message from any one of many ports at once. Returns the index of the /// port whose data is ready. (If multiple are ready, returns the lowest index.) pub fn select<A: Select>(ports: &mut [A]) -> uint { @@ -56,11 +66,13 @@ pub fn select<A: Select>(ports: &mut [A]) -> uint { // after letting the task get woken up. The and_then closure needs to delay // the task from resuming until all ports have become blocked_on. let (p,c) = comm::oneshot(); - let p = Cell::new(p); - let c = Cell::new(c); - (|| { - let c = Cell::new(c.take()); + { + let _guard = PortGuard { + port: Some(p), + }; + + let mut c = Some(c); let sched: ~Scheduler = Local::take(); sched.deschedule_running_task_and_then(|sched, task| { let task_handles = task.make_selectable(ports.len()); @@ -74,15 +86,12 @@ pub fn select<A: Select>(ports: &mut [A]) -> uint { } } - let c = Cell::new(c.take()); - do sched.event_loop.callback { c.take().send_deferred(()) } + let c = c.take_unwrap(); + do sched.event_loop.callback { + c.send_deferred(()) + } }) - }).finally(|| { - // Unkillable is necessary not because getting killed is dangerous here, - // but to force the recv not to use the same kill-flag that we used for - // selecting. Otherwise a user-sender could spuriously wakeup us here. - p.take().recv(); - }); + } // Task resumes. Now unblock ourselves from all the ports we blocked on. // If the success index wasn't reset, 'take' will just take all of them. @@ -133,7 +142,6 @@ mod test { use vec::*; use comm::GenericChan; use task; - use cell::Cell; use iter::{Iterator, range}; #[test] #[should_fail] @@ -246,9 +254,7 @@ mod test { let (p3,c3) = oneshot(); let (p4,c4) = oneshot(); - let x = Cell::new((c2, p3, c4)); do task::spawn { - let (c2, p3, c4) = x.take(); p3.recv(); // handshake parent c4.send(()); // normal receive task::deschedule(); @@ -284,10 +290,9 @@ mod test { let (p,c) = oneshot(); ports.push(p); if send_on_chans.contains(&i) { - let c = Cell::new(c); do spawntask_random { task::deschedule(); - c.take().send(()); + c.send(()); } } } diff --git a/src/libstd/task/mod.rs b/src/libstd/task/mod.rs index a587515bb16..24a24f24818 100644 --- a/src/libstd/task/mod.rs +++ b/src/libstd/task/mod.rs @@ -55,7 +55,6 @@ use prelude::*; -use cell::Cell; use comm::{stream, Chan, GenericChan, GenericPort, Port, Peekable}; use result::{Result, Ok, Err}; use rt::in_green_task_context; @@ -284,10 +283,8 @@ impl TaskBuilder { f } }; - let prev_gen_body = Cell::new(prev_gen_body); let next_gen_body = { let f: proc(proc()) -> proc() = proc(body) { - let prev_gen_body = prev_gen_body.take(); wrapper(prev_gen_body(body)) }; f @@ -432,12 +429,11 @@ pub fn with_task_name<U>(blk: |Option<&str>| -> U) -> U { use rt::task::Task; if in_green_task_context() { - Local::borrow(|task: &mut Task| { - match task.name { - Some(ref name) => blk(Some(name.as_slice())), - None => blk(None) - } - }) + let mut task = Local::borrow(None::<Task>); + match task.get().name { + Some(ref name) => blk(Some(name.as_slice())), + None => blk(None) + } } else { fail!("no task name exists in non-green task context") } @@ -459,7 +455,8 @@ pub fn failing() -> bool { use rt::task::Task; - Local::borrow(|local: &mut Task| local.unwinder.unwinding) + let mut local = Local::borrow(None::<Task>); + local.get().unwinder.unwinding } // The following 8 tests test the following 2^3 combinations: @@ -548,11 +545,9 @@ struct Wrapper { fn test_add_wrapper() { let (po, ch) = stream::<()>(); let mut b0 = task(); - let ch = Cell::new(ch); do b0.add_wrapper |body| { - let ch = Cell::new(ch.take()); + let ch = ch; let result: proc() = proc() { - let ch = ch.take(); body(); ch.send(()); }; @@ -606,9 +601,9 @@ fn test_try_fail() { #[cfg(test)] fn get_sched_id() -> int { - Local::borrow(|sched: &mut ::rt::sched::Scheduler| { - sched.sched_id() as int - }) + use rt::sched::Scheduler; + let mut sched = Local::borrow(None::<Scheduler>); + sched.get().sched_id() as int } #[test] @@ -642,12 +637,10 @@ fn test_spawn_sched_childs_on_default_sched() { // Assuming tests run on the default scheduler let default_id = get_sched_id(); - let ch = Cell::new(ch); do spawn_sched(SingleThreaded) { let parent_sched_id = get_sched_id(); - let ch = Cell::new(ch.take()); + let ch = ch; do spawn { - let ch = ch.take(); let child_sched_id = get_sched_id(); assert!(parent_sched_id != child_sched_id); assert_eq!(child_sched_id, default_id); @@ -671,10 +664,10 @@ fn test_spawn_sched_blocking() { let (fin_po, fin_ch) = stream(); let mut lock = Mutex::new(); - let lock2 = Cell::new(lock.clone()); + let lock2 = lock.clone(); do spawn_sched(SingleThreaded) { - let mut lock = lock2.take(); + let mut lock = lock2; lock.lock(); start_ch.send(()); diff --git a/src/libstd/task/spawn.rs b/src/libstd/task/spawn.rs index 153b3e4ce25..4ab7b74d300 100644 --- a/src/libstd/task/spawn.rs +++ b/src/libstd/task/spawn.rs @@ -77,7 +77,6 @@ use prelude::*; -use cell::Cell; use comm::{GenericChan, oneshot}; use rt::local::Local; use rt::sched::{Scheduler, Shutdown, TaskFromFriend}; @@ -134,23 +133,19 @@ pub fn spawn_raw(mut opts: TaskOpts, f: proc()) { // Create a task that will later be used to join with the new scheduler // thread when it is ready to terminate let (thread_port, thread_chan) = oneshot(); - let thread_port_cell = Cell::new(thread_port); let join_task = do Task::build_child(None) { debug!("running join task"); - let thread_port = thread_port_cell.take(); let thread: Thread<()> = thread_port.recv(); thread.join(); }; // Put the scheduler into another thread - let new_sched_cell = Cell::new(new_sched); - let orig_sched_handle_cell = Cell::new((*sched).make_handle()); - let join_task_cell = Cell::new(join_task); + let orig_sched_handle = (*sched).make_handle(); + let new_sched = new_sched; let thread = do Thread::start { - let mut new_sched = new_sched_cell.take(); - let mut orig_sched_handle = orig_sched_handle_cell.take(); - let join_task = join_task_cell.take(); + let mut new_sched = new_sched; + let mut orig_sched_handle = orig_sched_handle; let bootstrap_task = ~do Task::new_root(&mut new_sched.stack_pool, None) || { debug!("boostrapping a 1:1 scheduler"); @@ -178,9 +173,8 @@ pub fn spawn_raw(mut opts: TaskOpts, f: proc()) { if opts.notify_chan.is_some() { let notify_chan = opts.notify_chan.take_unwrap(); - let notify_chan = Cell::new(notify_chan); let on_exit: proc(UnwindResult) = proc(task_result) { - notify_chan.take().send(task_result) + notify_chan.send(task_result) }; task.death.on_exit = Some(on_exit); } diff --git a/src/libstd/unstable/dynamic_lib.rs b/src/libstd/unstable/dynamic_lib.rs index 2a6e40dc3a0..42a696eaf7e 100644 --- a/src/libstd/unstable/dynamic_lib.rs +++ b/src/libstd/unstable/dynamic_lib.rs @@ -140,7 +140,7 @@ pub mod dl { use path; use ptr; use str; - use unstable::sync::atomically; + use unstable::sync::atomic; use result::*; pub unsafe fn open_external(filename: &path::Path) -> *libc::c_void { @@ -158,25 +158,24 @@ pub mod dl { static mut lock: Mutex = MUTEX_INIT; unsafe { // dlerror isn't thread safe, so we need to lock around this entire - // sequence. `atomically` asserts that we don't do anything that + // sequence. `atomic` asserts that we don't do anything that // would cause this task to be descheduled, which could deadlock // the scheduler if it happens while the lock is held. // FIXME #9105 use a Rust mutex instead of C++ mutexes. - atomically(|| { - lock.lock(); - let _old_error = dlerror(); - - let result = f(); - - let last_error = dlerror(); - let ret = if ptr::null() == last_error { - Ok(result) - } else { - Err(str::raw::from_c_str(last_error)) - }; - lock.unlock(); - ret - }) + let _guard = atomic(); + lock.lock(); + let _old_error = dlerror(); + + let result = f(); + + let last_error = dlerror(); + let ret = if ptr::null() == last_error { + Ok(result) + } else { + Err(str::raw::from_c_str(last_error)) + }; + lock.unlock(); + ret } } @@ -209,7 +208,7 @@ pub mod dl { use libc; use path; use ptr; - use unstable::sync::atomically; + use unstable::sync::atomic; use result::*; pub unsafe fn open_external(filename: &path::Path) -> *libc::c_void { @@ -226,18 +225,17 @@ pub mod dl { pub fn check_for_errors_in<T>(f: || -> T) -> Result<T, ~str> { unsafe { - atomically(|| { - SetLastError(0); + let _guard = atomic(); + SetLastError(0); - let result = f(); + let result = f(); - let error = os::errno(); - if 0 == error { - Ok(result) - } else { - Err(format!("Error code {}", error)) - } - }) + let error = os::errno(); + if 0 == error { + Ok(result) + } else { + Err(format!("Error code {}", error)) + } } } diff --git a/src/libstd/unstable/mod.rs b/src/libstd/unstable/mod.rs index 63528036ffa..f8e2ea54f44 100644 --- a/src/libstd/unstable/mod.rs +++ b/src/libstd/unstable/mod.rs @@ -37,15 +37,13 @@ The executing thread has no access to a task pointer and will be using a normal large stack. */ pub fn run_in_bare_thread(f: proc()) { - use cell::Cell; use rt::thread::Thread; - let f_cell = Cell::new(f); let (port, chan) = comm::stream(); // FIXME #4525: Unfortunate that this creates an extra scheduler but it's // necessary since rust_raw_thread_join is blocking do task::spawn_sched(task::SingleThreaded) { - Thread::start(f_cell.take()).join(); + Thread::start(f).join(); chan.send(()); } port.recv(); diff --git a/src/libstd/unstable/sync.rs b/src/libstd/unstable/sync.rs index d8e437fda81..2dd5515bdbc 100644 --- a/src/libstd/unstable/sync.rs +++ b/src/libstd/unstable/sync.rs @@ -9,13 +9,11 @@ // except according to those terms. use cast; -use cell::Cell; use comm; use ptr; use option::{Option,Some,None}; use task; use unstable::atomics::{AtomicOption,AtomicUint,Acquire,Release,Relaxed,SeqCst}; -use unstable::finally::Finally; use unstable::mutex::Mutex; use ops::Drop; use clone::Clone; @@ -70,6 +68,35 @@ unsafe fn new_inner<T: Send>(data: T, refcount: uint) -> *mut ArcData<T> { cast::transmute(data) } +/// A helper object used by `UnsafeArc::unwrap`. +struct ChannelAndDataGuard<T> { + channel: Option<comm::ChanOne<bool>>, + data: Option<~ArcData<T>>, +} + +#[unsafe_destructor] +impl<T> Drop for ChannelAndDataGuard<T> { + fn drop(&mut self) { + if task::failing() { + // Killed during wait. Because this might happen while + // someone else still holds a reference, we can't free + // the data now; the "other" last refcount will free it. + unsafe { + let channel = self.channel.take_unwrap(); + let data = self.data.take_unwrap(); + channel.send(false); + cast::forget(data); + } + } + } +} + +impl<T> ChannelAndDataGuard<T> { + fn unwrap(mut self) -> (comm::ChanOne<bool>, ~ArcData<T>) { + (self.channel.take_unwrap(), self.data.take_unwrap()) + } +} + impl<T: Send> UnsafeArc<T> { pub fn new(data: T) -> UnsafeArc<T> { unsafe { UnsafeArc { data: new_inner(data, 1) } } @@ -160,32 +187,19 @@ impl<T: Send> UnsafeArc<T> { data.data.take_unwrap() } else { // The *next* person who sees the refcount hit 0 will wake us. - let p1 = Cell::new(p1); // argh - // Unlike the above one, this cell is necessary. It will get - // taken either in the do block or in the finally block. - let c2_and_data = Cell::new((c2,data)); - (|| { - p1.take().recv(); - // Got here. Back in the 'unkillable' without getting killed. - let (c2, data) = c2_and_data.take(); - c2.send(true); - // FIXME(#3224): it should be like this - // let ~ArcData { data: user_data, _ } = data; - // user_data - let mut data = data; - data.data.take_unwrap() - }).finally(|| { - if task::failing() { - // Killed during wait. Because this might happen while - // someone else still holds a reference, we can't free - // the data now; the "other" last refcount will free it. - let (c2, data) = c2_and_data.take(); - c2.send(false); - cast::forget(data); - } else { - assert!(c2_and_data.is_empty()); - } - }) + let c2_and_data = ChannelAndDataGuard { + channel: Some(c2), + data: Some(data), + }; + p1.recv(); + // Got here. Back in the 'unkillable' without getting killed. + let (c2, data) = c2_and_data.unwrap(); + c2.send(true); + // FIXME(#3224): it should be like this + // let ~ArcData { data: user_data, _ } = data; + // user_data + let mut data = data; + data.data.take_unwrap() } } else { // If 'put' returns the server end back to us, we were rejected; @@ -280,17 +294,44 @@ impl<T> Drop for UnsafeArc<T>{ /****************************************************************************/ +pub struct AtomicGuard { + on: bool, +} + +impl Drop for AtomicGuard { + fn drop(&mut self) { + use rt::task::{Task, GreenTask, SchedTask}; + use rt::local::Local; + + if self.on { + unsafe { + let task_opt: Option<*mut Task> = Local::try_unsafe_borrow(); + match task_opt { + Some(t) => { + match (*t).task_type { + GreenTask(_) => (*t).death.allow_deschedule(), + SchedTask => {} + } + } + None => {} + } + } + } + } +} + /** - * Enables a runtime assertion that no operation in the argument closure shall - * use scheduler operations (deschedule, recv, spawn, etc). This is for use with - * pthread mutexes, which may block the entire scheduler thread, rather than - * just one task, and is hence prone to deadlocks if mixed with descheduling. + * Enables a runtime assertion that no operation while the returned guard is + * live uses scheduler operations (deschedule, recv, spawn, etc). This is for + * use with pthread mutexes, which may block the entire scheduler thread, + * rather than just one task, and is hence prone to deadlocks if mixed with + * descheduling. * * NOTE: THIS DOES NOT PROVIDE LOCKING, or any sort of critical-section * synchronization whatsoever. It only makes sense to use for CPU-local issues. */ // FIXME(#8140) should not be pub -pub unsafe fn atomically<U>(f: || -> U) -> U { +pub unsafe fn atomic() -> AtomicGuard { use rt::task::{Task, GreenTask, SchedTask}; use rt::local::Local; @@ -299,15 +340,19 @@ pub unsafe fn atomically<U>(f: || -> U) -> U { Some(t) => { match (*t).task_type { GreenTask(_) => { - (|| { - (*t).death.inhibit_deschedule(); - f() - }).finally(|| (*t).death.allow_deschedule()) + (*t).death.inhibit_deschedule(); + return AtomicGuard { + on: true, + }; } - SchedTask => f() + SchedTask => {} } } - None => f() + None => {} + } + + AtomicGuard { + on: false, } } @@ -466,7 +511,7 @@ mod tests { use comm; use option::*; use prelude::*; - use super::{Exclusive, UnsafeArc, atomically}; + use super::{Exclusive, UnsafeArc, atomic}; use task; use mem::size_of; @@ -478,10 +523,10 @@ mod tests { } #[test] - fn test_atomically() { + fn test_atomic() { // NB. The whole runtime will abort on an 'atomic-sleep' violation, // so we can't really test for the converse behaviour. - unsafe { atomically(|| ()) } task::deschedule(); // oughtn't fail + unsafe { let _ = atomic(); } // oughtn't fail } #[test] |