nordic-dev.net/rust
nordic-dev.net/rust
2
0
Fork 0
mirror of https://github.com/rust-lang/rust.git synced 2024-11-23 23:34:48 +00:00
Code Issues Packages Projects Releases Wiki Activity

auto merge of #10791 : pcwalton/rust/decelling, r=pcwalton

Browse Source 
34 uses of `Cell` remain.

r? @alexcrichton
This commit is contained in:
bors 2013-12-10 19:16:19 -08:00
commit b8b16ae099
70 changed files with 873 additions and 964 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

14
src/compiletest/compiletest.rs
View File

@ -325,19 +325,17 @@ pub fn make_test_name(config: &config, testfile: &Path) -> test::TestName {
}
pub fn make_test_closure(config: &config, testfile: &Path) -> test::TestFn {
use std::cell::Cell;
let config = Cell::new((*config).clone());
let config = (*config).clone();
// FIXME (#9639): This needs to handle non-utf8 paths
let testfile = Cell::new(testfile.as_str().unwrap().to_owned());
test::DynTestFn(proc() { runtest::run(config.take(), testfile.take()) })
let testfile = testfile.as_str().unwrap().to_owned();
test::DynTestFn(proc() { runtest::run(config, testfile) })
}
pub fn make_metrics_test_closure(config: &config, testfile: &Path) -> test::TestFn {
use std::cell::Cell;
let config = Cell::new((*config).clone());
let config = (*config).clone();
// FIXME (#9639): This needs to handle non-utf8 paths
let testfile = Cell::new(testfile.as_str().unwrap().to_owned());
let testfile = testfile.as_str().unwrap().to_owned();
test::DynMetricFn(proc(mm) {
runtest::run_metrics(config.take(), testfile.take(), mm)
runtest::run_metrics(config, testfile, mm)
})
}

9
src/libextra/arc.rs
View File

@ -597,7 +597,6 @@ mod tests {
use arc::*;
use std::cell::Cell;
use std::comm;
use std::task;
@ -628,18 +627,18 @@ mod tests {
let arc = ~MutexArc::new(false);
let arc2 = ~arc.clone();
let (p,c) = comm::oneshot();
let (c,p) = (Cell::new(c), Cell::new(p));
do task::spawn || {
do task::spawn {
// wait until parent gets in
p.take().recv();
p.recv();
arc2.access_cond(|state, cond| {
*state = true;
cond.signal();
})
}
let mut c = Some(c);
arc.access_cond(|state, cond| {
c.take().send(());
c.take_unwrap().send(());
assert!(!*state);
while !*state {
cond.wait();

9
src/libextra/future.rs
View File

@ -25,7 +25,6 @@
#[allow(missing_doc)];
use std::cell::Cell;
use std::comm::{PortOne, oneshot};
use std::util::replace;
@ -113,9 +112,8 @@ impl<A:Send> Future<A> {
* waiting for the result to be received on the port.
*/
let port = Cell::new(port);
do Future::from_fn {
port.take().recv()
port.recv()
}
}
@ -141,7 +139,6 @@ impl<A:Send> Future<A> {
mod test {
use future::Future;
use std::cell::Cell;
use std::comm::oneshot;
use std::task;
@ -199,9 +196,9 @@ mod test {
#[test]
fn test_sendable_future() {
let expected = "schlorf";
let f = Cell::new(do Future::spawn { expected });
let f = do Future::spawn { expected };
do task::spawn {
let mut f = f.take();
let mut f = f;
let actual = f.get();
assert_eq!(actual, expected);
}

9
src/libextra/sync.rs
View File

@ -676,7 +676,6 @@ mod tests {
use sync::*;
use std::cast;
use std::cell::Cell;
use std::comm;
use std::result;
use std::task;
@ -762,9 +761,9 @@ mod tests {
let s = Semaphore::new(1);
let s2 = s.clone();
let (p, c) = comm::stream();
let child_data = Cell::new((s2, c));
let mut child_data = Some((s2, c));
s.access(|| {
let (s2, c) = child_data.take();
let (s2, c) = child_data.take_unwrap();
do task::spawn {
c.send(());
s2.access(|| { });
@ -947,13 +946,13 @@ mod tests {
let mut sibling_convos = ~[];
2.times(|| {
let (p, c) = comm::stream();
let c = Cell::new(c);
sibling_convos.push(p);
let mi = m2.clone();
// spawn sibling task
do task::spawn { // linked
let mut c = Some(c);
mi.lock_cond(|cond| {
let c = c.take();
let c = c.take_unwrap();
c.send(()); // tell sibling to go ahead
(|| {
cond.wait(); // block forever

3
src/libextra/test.rs
View File

@ -872,7 +872,6 @@ pub fn run_test(force_ignore: bool,
fn run_test_inner(desc: TestDesc,
monitor_ch: SharedChan<MonitorMsg>,
testfn: proc()) {
let testfn_cell = ::std::cell::Cell::new(testfn);
do task::spawn {
let mut task = task::task();
task.name(match desc.name {
@ -880,7 +879,7 @@ pub fn run_test(force_ignore: bool,
StaticTestName(name) => SendStrStatic(name),
});
let result_future = task.future_result();
task.spawn(testfn_cell.take());
task.spawn(testfn);
let task_result = result_future.recv();
let test_result = calc_result(&desc, task_result.is_ok());

3
src/libextra/workcache.rs
View File

@ -15,7 +15,6 @@ use json::ToJson;
use serialize::{Encoder, Encodable, Decoder, Decodable};
use arc::{Arc,RWArc};
use treemap::TreeMap;
use std::cell::Cell;
use std::comm::{PortOne, oneshot};
use std::{str, task};
use std::io;
@ -430,7 +429,6 @@ impl<'self> Prep<'self> {
debug!("Cache miss!");
let (port, chan) = oneshot();
let blk = bo.take_unwrap();
let chan = Cell::new(chan);
// XXX: What happens if the task fails?
do task::spawn {
@ -438,7 +436,6 @@ impl<'self> Prep<'self> {
discovered_inputs: WorkMap::new(),
discovered_outputs: WorkMap::new(),
};
let chan = chan.take();
let v = blk(&mut exe);
chan.send((exe, v));
}

39
src/librustdoc/html/render.rs
View File

@ -33,7 +33,6 @@
//! These tasks are not parallelized (they haven't been a bottleneck yet), and
//! both occur before the crate is rendered.
use std::cell::Cell;
use std::comm::{SharedPort, SharedChan};
use std::comm;
use std::fmt;
@ -46,7 +45,6 @@ use std::io::File;
use std::os;
use std::str;
use std::task;
use std::unstable::finally::Finally;
use std::vec;
use extra::arc::RWArc;
@ -642,6 +640,22 @@ impl<'self> Cache {
}
}
enum Progress {
JobNew,
JobDone,
}
/// A helper object to unconditionally send a value on a chanel.
struct ChannelGuard {
channel: SharedChan<Progress>,
}
impl Drop for ChannelGuard {
fn drop(&mut self) {
self.channel.send(JobDone)
}
}
impl Context {
/// Recurse in the directory structure and change the "root path" to make
/// sure it always points to the top (relatively)
@ -674,8 +688,6 @@ impl Context {
Die,
Process(Context, clean::Item),
}
enum Progress { JobNew, JobDone }
let workers = match os::getenv("RUSTDOC_WORKERS") {
Some(s) => {
match from_str::<uint>(s) {
@ -725,17 +737,16 @@ impl Context {
match port.recv() {
Process(cx, item) => {
let mut cx = cx;
let item = Cell::new(item);
(|| {
cx.item(item.take(), |cx, item| {
// If we fail, everything else should still get
// completed.
let _guard = ChannelGuard {
channel: prog_chan.clone(),
};
cx.item(item, |cx, item| {
prog_chan.send(JobNew);
chan.send(Process(cx.clone(), item));
})
}).finally(|| {
// If we fail, everything else should still get
// completed
prog_chan.send(JobDone);
})
}
Die => break,
}
@ -802,9 +813,9 @@ impl Context {
// recurse into the items of the module as well.
clean::ModuleItem(..) => {
let name = item.name.get_ref().to_owned();
let item = Cell::new(item);
let mut item = Some(item);
self.recurse(name, |this| {
let item = item.take();
let item = item.take_unwrap();
let dst = this.dst.join("index.html");
render(File::create(&dst).unwrap(), this, &item, false);

11
src/librustdoc/lib.rs
View File

@ -26,7 +26,6 @@ extern mod syntax;
extern mod rustc;
extern mod extra;
use std::cell::Cell;
use std::local_data;
use std::io;
use std::io::File;
@ -196,13 +195,13 @@ fn rust_input(cratefile: &str, matches: &getopts::Matches) -> Output {
let mut plugins = matches.opt_strs("plugins");
// First, parse the crate and extract all relevant information.
let libs = Cell::new(matches.opt_strs("L").map(|s| Path::new(s.as_slice())));
let cfgs = Cell::new(matches.opt_strs("cfg"));
let cr = Cell::new(Path::new(cratefile));
let libs = matches.opt_strs("L").map(|s| Path::new(s.as_slice()));
let cfgs = matches.opt_strs("cfg");
let cr = Path::new(cratefile);
info!("starting to run rustc");
let (crate, analysis) = do std::task::try {
let cr = cr.take();
core::run_core(libs.take().move_iter().collect(), cfgs.take(), &cr)
let cr = cr;
core::run_core(libs.move_iter().collect(), cfgs, &cr)
}.unwrap();
info!("finished with rustc");
local_data::set(analysiskey, analysis);

9
src/librustdoc/passes.rs
View File

@ -8,12 +8,10 @@
// option. This file may not be copied, modified, or distributed
// except according to those terms.
use std::num;
use std::cell::Cell;
use std::uint;
use std::hashmap::HashSet;
use std::local_data;
use std::num;
use std::uint;
use syntax::ast;
use clean;
@ -56,11 +54,10 @@ pub fn strip_hidden(crate: clean::Crate) -> plugins::PluginResult {
pub fn strip_private(crate: clean::Crate) -> plugins::PluginResult {
// This stripper collects all *retained* nodes.
let mut retained = HashSet::new();
let crate = Cell::new(crate);
let exported_items = local_data::get(super::analysiskey, |analysis| {
analysis.unwrap().exported_items.clone()
});
let mut crate = crate.take();
let mut crate = crate;
// strip all private items
{

15
src/librustpkg/installed_packages.rs
View File

@ -19,7 +19,10 @@ use std::io::fs;
pub fn list_installed_packages(f: |&PkgId| -> bool) -> bool {
let workspaces = rust_path();
for p in workspaces.iter() {
let binfiles = io::ignore_io_error(|| fs::readdir(&p.join("bin")));
let binfiles = {
let _guard = io::ignore_io_error();
fs::readdir(&p.join("bin"))
};
for exec in binfiles.iter() {
// FIXME (#9639): This needs to handle non-utf8 paths
match exec.filestem_str() {
@ -31,7 +34,10 @@ pub fn list_installed_packages(f: |&PkgId| -> bool) -> bool {
}
}
}
let libfiles = io::ignore_io_error(|| fs::readdir(&p.join("lib")));
let libfiles = {
let _guard = io::ignore_io_error();
fs::readdir(&p.join("lib"))
};
for lib in libfiles.iter() {
debug!("Full name: {}", lib.display());
match has_library(lib) {
@ -55,7 +61,10 @@ pub fn list_installed_packages(f: |&PkgId| -> bool) -> bool {
}
pub fn has_library(p: &Path) -> Option<~str> {
let files = io::ignore_io_error(|| fs::readdir(p));
let files = {
let _guard = io::ignore_io_error();
fs::readdir(p)
};
for path in files.iter() {
if path.extension_str() == Some(os::consts::DLL_EXTENSION) {
let stuff : &str = path.filestem_str().expect("has_library: weird path");

5
src/librustpkg/path_util.rs
View File

@ -217,7 +217,10 @@ pub fn system_library(sysroot: &Path, lib_name: &str) -> Option<Path> {
fn library_in(short_name: &str, version: &Version, dir_to_search: &Path) -> Option<Path> {
debug!("Listing directory {}", dir_to_search.display());
let dir_contents = io::ignore_io_error(|| fs::readdir(dir_to_search));
let dir_contents = {
let _guard = io::ignore_io_error();
fs::readdir(dir_to_search)
};
debug!("dir has {:?} entries", dir_contents.len());
let lib_prefix = format!("{}{}", os::consts::DLL_PREFIX, short_name);

6
src/librustuv/async.rs
View File

@ -125,7 +125,6 @@ impl Drop for AsyncWatcher {
#[cfg(test)]
mod test_remote {
use std::cell::Cell;
use std::rt::rtio::Callback;
use std::rt::thread::Thread;
use std::rt::tube::Tube;
@ -150,10 +149,11 @@ mod test_remote {
let mut tube = Tube::new();
let cb = ~MyCallback(Some(tube.clone()));
let watcher = Cell::new(AsyncWatcher::new(local_loop(), cb as ~Callback));
let watcher = AsyncWatcher::new(local_loop(), cb as ~Callback);
let thread = do Thread::start {
watcher.take().fire();
let mut watcher = watcher;
watcher.fire();
};
assert_eq!(tube.recv(), 1);

23
src/librustuv/lib.rs
View File

@ -164,16 +164,20 @@ pub struct ForbidSwitch {
impl ForbidSwitch {
fn new(s: &'static str) -> ForbidSwitch {
let mut sched = Local::borrow(None::<Scheduler>);
ForbidSwitch {
msg: s, sched: Local::borrow(|s: &mut Scheduler| s.sched_id())
msg: s,
sched: sched.get().sched_id(),
}
}
}
impl Drop for ForbidSwitch {
fn drop(&mut self) {
assert!(self.sched == Local::borrow(|s: &mut Scheduler| s.sched_id()),
"didnt want a scheduler switch: {}", self.msg);
let mut sched = Local::borrow(None::<Scheduler>);
assert!(self.sched == sched.get().sched_id(),
"didnt want a scheduler switch: {}",
self.msg);
}
}
@ -391,15 +395,12 @@ pub fn slice_to_uv_buf(v: &[u8]) -> Buf {
#[cfg(test)]
fn local_loop() -> &'static mut Loop {
unsafe {
cast::transmute(Local::borrow(|sched: &mut Scheduler| {
let mut io = None;
sched.event_loop.io(|i| {
cast::transmute({
let mut sched = Local::borrow(None::<Scheduler>);
let (_vtable, uvio): (uint, &'static mut uvio::UvIoFactory) =
cast::transmute(i);
io = Some(uvio);
});
io.unwrap()
}).uv_loop())
cast::transmute(sched.get().event_loop.io().unwrap());
uvio
}.uv_loop())
}
}

5
src/librustuv/macros.rs
View File

@ -29,7 +29,10 @@ macro_rules! uvdebug (
// get a handle for the current scheduler
macro_rules! get_handle_to_current_scheduler(
() => (Local::borrow(|sched: &mut Scheduler| sched.make_handle()))
() => ({
let mut sched = Local::borrow(None::<Scheduler>);
sched.get().make_handle()
})
)
pub fn dumb_println(args: &fmt::Arguments) {

104
src/librustuv/net.rs
View File

@ -646,7 +646,6 @@ impl Drop for UdpWatcher {
#[cfg(test)]
mod test {
use std::cell::Cell;
use std::comm::oneshot;
use std::rt::test::*;
use std::rt::rtio::{RtioTcpStream, RtioTcpListener, RtioTcpAcceptor,
@ -691,7 +690,6 @@ mod test {
#[test]
fn listen_ip4() {
let (port, chan) = oneshot();
let chan = Cell::new(chan);
let addr = next_test_ip4();
do spawn {
@ -701,7 +699,7 @@ mod test {
let mut w = match w.listen() {
Ok(w) => w, Err(e) => fail!("{:?}", e),
};
chan.take().send(());
chan.send(());
match w.accept() {
Ok(mut stream) => {
let mut buf = [0u8, ..10];
@ -728,7 +726,6 @@ mod test {
#[test]
fn listen_ip6() {
let (port, chan) = oneshot();
let chan = Cell::new(chan);
let addr = next_test_ip6();
do spawn {
@ -738,7 +735,7 @@ mod test {
let mut w = match w.listen() {
Ok(w) => w, Err(e) => fail!("{:?}", e),
};
chan.take().send(());
chan.send(());
match w.accept() {
Ok(mut stream) => {
let mut buf = [0u8, ..10];
@ -765,14 +762,13 @@ mod test {
#[test]
fn udp_recv_ip4() {
let (port, chan) = oneshot();
let chan = Cell::new(chan);
let client = next_test_ip4();
let server = next_test_ip4();
do spawn {
match UdpWatcher::bind(local_loop(), server) {
Ok(mut w) => {
chan.take().send(());
chan.send(());
let mut buf = [0u8, ..10];
match w.recvfrom(buf) {
Ok((10, addr)) => assert_eq!(addr, client),
@ -798,14 +794,13 @@ mod test {
#[test]
fn udp_recv_ip6() {
let (port, chan) = oneshot();
let chan = Cell::new(chan);
let client = next_test_ip6();
let server = next_test_ip6();
do spawn {
match UdpWatcher::bind(local_loop(), server) {
Ok(mut w) => {
chan.take().send(());
chan.send(());
let mut buf = [0u8, ..10];
match w.recvfrom(buf) {
Ok((10, addr)) => assert_eq!(addr, client),
@ -834,13 +829,11 @@ mod test {
let addr = next_test_ip4();
static MAX: uint = 5000;
let (port, chan) = oneshot();
let port = Cell::new(port);
let chan = Cell::new(chan);
do spawn {
let listener = TcpListener::bind(local_loop(), addr).unwrap();
let mut acceptor = listener.listen().unwrap();
chan.take().send(());
chan.send(());
let mut stream = acceptor.accept().unwrap();
let buf = [1, .. 2048];
let mut total_bytes_written = 0;
@ -852,7 +845,7 @@ mod test {
}
do spawn {
port.take().recv();
port.recv();
let mut stream = TcpWatcher::connect(local_loop(), addr).unwrap();
let mut buf = [0, .. 2048];
let mut total_bytes_read = 0;
@ -873,18 +866,16 @@ mod test {
let server_addr = next_test_ip4();
let client_addr = next_test_ip4();
let (port, chan) = oneshot();
let port = Cell::new(port);
let chan = Cell::new(chan);
do spawn {
let mut client = UdpWatcher::bind(local_loop(), client_addr).unwrap();
port.take().recv();
port.recv();
assert!(client.sendto([1], server_addr).is_ok());
assert!(client.sendto([2], server_addr).is_ok());
}
let mut server = UdpWatcher::bind(local_loop(), server_addr).unwrap();
chan.take().send(());
chan.send(());
let mut buf1 = [0];
let mut buf2 = [0];
let (nread1, src1) = server.recvfrom(buf1).unwrap();
@ -908,14 +899,11 @@ mod test {
let (p1, c1) = oneshot();
let (p2, c2) = oneshot();
let first = Cell::new((p1, c2));
let second = Cell::new((p2, c1));
do spawn {
let l = local_loop();
let mut server_out = UdpWatcher::bind(l, server_out_addr).unwrap();
let mut server_in = UdpWatcher::bind(l, server_in_addr).unwrap();
let (port, chan) = first.take();
let (port, chan) = (p1, c2);
chan.send(());
port.recv();
let msg = [1, .. 2048];
@ -939,7 +927,7 @@ mod test {
let l = local_loop();
let mut client_out = UdpWatcher::bind(l, client_out_addr).unwrap();
let mut client_in = UdpWatcher::bind(l, client_in_addr).unwrap();
let (port, chan) = second.take();
let (port, chan) = (p2, c1);
port.recv();
chan.send(());
let mut total_bytes_recv = 0;
@ -966,14 +954,12 @@ mod test {
fn test_read_and_block() {
let addr = next_test_ip4();
let (port, chan) = oneshot();
let port = Cell::new(port);
let chan = Cell::new(chan);
do spawn {
let listener = TcpListener::bind(local_loop(), addr).unwrap();
let mut acceptor = listener.listen().unwrap();
let (port2, chan2) = stream();
chan.take().send(port2);
chan.send(port2);
let mut stream = acceptor.accept().unwrap();
let mut buf = [0, .. 2048];
@ -998,7 +984,7 @@ mod test {
}
do spawn {
let port2 = port.take().recv();
let port2 = port.recv();
let mut stream = TcpWatcher::connect(local_loop(), addr).unwrap();
stream.write([0, 1, 2, 3, 4, 5, 6, 7]);
stream.write([0, 1, 2, 3, 4, 5, 6, 7]);
@ -1041,18 +1027,14 @@ mod test {
#[test]
fn test_homing_closes_correctly() {
let (port, chan) = oneshot();
let port = Cell::new(port);
let chan = Cell::new(chan);
do task::spawn_sched(task::SingleThreaded) {
let chan = Cell::new(chan.take());
let listener = UdpWatcher::bind(local_loop(), next_test_ip4()).unwrap();
chan.take().send(listener);
chan.send(listener);
}
do task::spawn_sched(task::SingleThreaded) {
let port = Cell::new(port.take());
port.take().recv();
port.recv();
}
}
@ -1086,22 +1068,22 @@ mod test {
let mut sched2 = ~Scheduler::new(loop2, worker2, queues.clone(),
sleepers.clone());
let handle1 = Cell::new(sched1.make_handle());
let handle2 = Cell::new(sched2.make_handle());
let tasksFriendHandle = Cell::new(sched2.make_handle());
let handle1 = sched1.make_handle();
let handle2 = sched2.make_handle();
let tasksFriendHandle = sched2.make_handle();
let on_exit: proc(UnwindResult) = proc(exit_status) {
handle1.take().send(Shutdown);
handle2.take().send(Shutdown);
let mut handle1 = handle1;
let mut handle2 = handle2;
handle1.send(Shutdown);
handle2.send(Shutdown);
assert!(exit_status.is_success());
};
unsafe fn local_io() -> &'static mut IoFactory {
Local::borrow(|sched: &mut Scheduler| {
let mut io = None;
sched.event_loop.io(|i| io = Some(i));
let mut sched = Local::borrow(None::<Scheduler>);
let io = sched.get().event_loop.io();
cast::transmute(io.unwrap())
})
}
let test_function: proc() = proc() {
@ -1113,11 +1095,13 @@ mod test {
// block self on sched1
let scheduler: ~Scheduler = Local::take();
let mut tasksFriendHandle = Some(tasksFriendHandle);
scheduler.deschedule_running_task_and_then(|_, task| {
// unblock task
task.wake().map(|task| {
// send self to sched2
tasksFriendHandle.take().send(TaskFromFriend(task));
tasksFriendHandle.take_unwrap()
.send(TaskFromFriend(task));
});
// sched1 should now sleep since it has nothing else to do
})
@ -1133,19 +1117,20 @@ mod test {
let mut main_task = ~Task::new_root(&mut sched1.stack_pool, None,
test_function);
main_task.death.on_exit = Some(on_exit);
let main_task = Cell::new(main_task);
let null_task = Cell::new(~do Task::new_root(&mut sched2.stack_pool,
None) || {});
let sched1 = Cell::new(sched1);
let sched2 = Cell::new(sched2);
let thread1 = do Thread::start {
sched1.take().bootstrap(main_task.take());
let null_task = ~do Task::new_root(&mut sched2.stack_pool, None) {
// nothing
};
let main_task = main_task;
let sched1 = sched1;
let thread1 = do Thread::start {
sched1.bootstrap(main_task);
};
let sched2 = sched2;
let thread2 = do Thread::start {
sched2.take().bootstrap(null_task.take());
sched2.bootstrap(null_task);
};
thread1.join();
@ -1164,13 +1149,12 @@ mod test {
#[should_fail] #[test]
fn tcp_stream_fail_cleanup() {
let (port, chan) = oneshot();
let chan = Cell::new(chan);
let addr = next_test_ip4();
do spawn {
let w = TcpListener::bind(local_loop(), addr).unwrap();
let mut w = w.listen().unwrap();
chan.take().send(());
chan.send(());
w.accept();
}
port.recv();
@ -1189,14 +1173,13 @@ mod test {
fn udp_fail_other_task() {
let addr = next_test_ip4();
let (port, chan) = oneshot();
let chan = Cell::new(chan);
// force the handle to be created on a different scheduler, failure in
// the original task will force a homing operation back to this
// scheduler.
do task::spawn_sched(task::SingleThreaded) {
let w = UdpWatcher::bind(local_loop(), addr).unwrap();
chan.take().send(w);
chan.send(w);
}
let _w = port.recv();
@ -1208,13 +1191,12 @@ mod test {
#[ignore(reason = "linked failure")]
fn linked_failure1() {
let (port, chan) = oneshot();
let chan = Cell::new(chan);
let addr = next_test_ip4();
do spawn {
let w = TcpListener::bind(local_loop(), addr).unwrap();
let mut w = w.listen().unwrap();
chan.take().send(());
chan.send(());
w.accept();
}
@ -1227,13 +1209,12 @@ mod test {
#[ignore(reason = "linked failure")]
fn linked_failure2() {
let (port, chan) = oneshot();
let chan = Cell::new(chan);
let addr = next_test_ip4();
do spawn {
let w = TcpListener::bind(local_loop(), addr).unwrap();
let mut w = w.listen().unwrap();
chan.take().send(());
chan.send(());
let mut buf = [0];
w.accept().unwrap().read(buf);
}
@ -1249,11 +1230,10 @@ mod test {
#[ignore(reason = "linked failure")]
fn linked_failure3() {
let (port, chan) = stream();
let chan = Cell::new(chan);
let addr = next_test_ip4();
do spawn {
let chan = chan.take();
let chan = chan;
let w = TcpListener::bind(local_loop(), addr).unwrap();
let mut w = w.listen().unwrap();
chan.send(());

7
src/librustuv/pipe.rs
View File

@ -231,7 +231,6 @@ impl HomingIO for PipeAcceptor {
#[cfg(test)]
mod tests {
use std::cell::Cell;
use std::comm::oneshot;
use std::rt::rtio::{RtioUnixListener, RtioUnixAcceptor, RtioPipe};
use std::rt::test::next_test_unix;
@ -276,12 +275,11 @@ mod tests {
let path = next_test_unix();
let path2 = path.clone();
let (port, chan) = oneshot();
let chan = Cell::new(chan);
do spawn {
let p = PipeListener::bind(local_loop(), &path2.to_c_str()).unwrap();
let mut p = p.listen().unwrap();
chan.take().send(());
chan.send(());
let mut client = p.accept().unwrap();
let mut buf = [0];
assert!(client.read(buf).unwrap() == 1);
@ -301,12 +299,11 @@ mod tests {
let path = next_test_unix();
let path2 = path.clone();
let (port, chan) = oneshot();
let chan = Cell::new(chan);
do spawn {
let p = PipeListener::bind(local_loop(), &path2.to_c_str()).unwrap();
let mut p = p.listen().unwrap();
chan.take().send(());
chan.send(());
p.accept();
}
port.recv();

4
src/librustuv/signal.rs
View File

@ -76,7 +76,6 @@ impl Drop for SignalWatcher {
#[cfg(test)]
mod test {
use super::*;
use std::cell::Cell;
use super::super::local_loop;
use std::io::signal;
use std::comm::{SharedChan, stream};
@ -89,9 +88,8 @@ mod test {
let _signal = SignalWatcher::new(local_loop(), signal::Interrupt,
chan);
let port = Cell::new(port);
do spawn {
port.take().try_recv();
port.try_recv();
}
// when we drop the SignalWatcher we're going to destroy the channel,

13
src/librustuv/timer.rs
View File

@ -163,7 +163,6 @@ impl Drop for TimerWatcher {
#[cfg(test)]
mod test {
use super::*;
use std::cell::Cell;
use std::rt::rtio::RtioTimer;
use super::super::local_loop;
@ -229,10 +228,10 @@ mod test {
fn closing_channel_during_drop_doesnt_kill_everything() {
// see issue #10375
let mut timer = TimerWatcher::new(local_loop());
let timer_port = Cell::new(timer.period(1000));
let timer_port = timer.period(1000);
do spawn {
timer_port.take().try_recv();
timer_port.try_recv();
}
// when we drop the TimerWatcher we're going to destroy the channel,
@ -243,10 +242,10 @@ mod test {
fn reset_doesnt_switch_tasks() {
// similar test to the one above.
let mut timer = TimerWatcher::new(local_loop());
let timer_port = Cell::new(timer.period(1000));
let timer_port = timer.period(1000);
do spawn {
timer_port.take().try_recv();
timer_port.try_recv();
}
timer.oneshot(1);
@ -255,10 +254,10 @@ mod test {
fn reset_doesnt_switch_tasks2() {
// similar test to the one above.
let mut timer = TimerWatcher::new(local_loop());
let timer_port = Cell::new(timer.period(1000));
let timer_port = timer.period(1000);
do spawn {
timer_port.take().try_recv();
timer_port.try_recv();
}
timer.sleep(1);

22
src/librustuv/uvio.rs
View File

@ -45,9 +45,10 @@ pub trait HomingIO {
let _f = ForbidUnwind::new("going home");
let current_sched_id = Local::borrow(|sched: &mut Scheduler| {
sched.sched_id()
});
let current_sched_id = {
let mut sched = Local::borrow(None::<Scheduler>);
sched.get().sched_id()
};
// Only need to invoke a context switch if we're not on the right
// scheduler.
@ -59,9 +60,10 @@ pub trait HomingIO {
});
})
}
let current_sched_id = Local::borrow(|sched: &mut Scheduler| {
sched.sched_id()
});
let current_sched_id = {
let mut sched = Local::borrow(None::<Scheduler>);
sched.get().sched_id()
};
assert!(current_sched_id == self.home().sched_id);
self.home().sched_id
@ -96,7 +98,8 @@ struct HomingMissile {
impl HomingMissile {
pub fn check(&self, msg: &'static str) {
let local_id = Local::borrow(|sched: &mut Scheduler| sched.sched_id());
let mut sched = Local::borrow(None::<Scheduler>);
let local_id = sched.get().sched_id();
assert!(local_id == self.io_home, "{}", msg);
}
}
@ -158,8 +161,9 @@ impl EventLoop for UvEventLoop {
~AsyncWatcher::new(self.uvio.uv_loop(), f) as ~RemoteCallback
}
fn io<'a>(&'a mut self, f: |&'a mut IoFactory|) {
f(&mut self.uvio as &mut IoFactory)
fn io<'a>(&'a mut self) -> Option<&'a mut IoFactory> {
let factory = &mut self.uvio as &mut IoFactory;
Some(factory)
}
}

6
src/libstd/at_vec.rs
View File

@ -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 *()
}
}

60
src/libstd/cell.rs
View File

@ -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> {

20
src/libstd/condition.rs
View File

@ -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>
}

12
src/libstd/io/fs.rs
View File

@ -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) {
let mut io = LocalIo::borrow();
match f(io.get()) {
Ok(t) => Some(t),
Err(ioerr) => {
io_error::cond.raise(ioerr);
None
}
}
})
}
impl File {
@ -132,8 +131,8 @@ 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) {
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,
@ -144,7 +143,6 @@ impl File {
None
}
}
})
}
/// Attempts to open a file in read-only mode. This function is equivalent to

5
src/libstd/io/mod.rs
View File

@ -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

3
src/libstd/io/native/mod.rs
View File

@ -223,3 +223,6 @@ impl rtio::IoFactory for IoFactory {
Err(unimpl())
}
}
pub static mut NATIVE_IO_FACTORY: IoFactory = IoFactory;

11
src/libstd/io/net/addrinfo.rs
View File

@ -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) {
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)]

108
src/libstd/io/net/tcp.rs
View File

@ -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) {
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) {
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());

32
src/libstd/io/net/udp.rs
View File

@ -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) {
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!()

34
src/libstd/io/net/unix.rs
View File

@ -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()) {
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()) {
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]);

7
src/libstd/io/pipe.rs
View File

@ -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) {
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 {

9
src/libstd/io/process.rs
View File

@ -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,9 +119,8 @@ 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()) {
let mut io = LocalIo::borrow();
match io.get().spawn(config) {
Ok((p, io)) => Some(Process{
handle: p,
io: io.move_iter().map(|p|
@ -134,7 +132,6 @@ impl Process {
None
}
}
})
}
/// Returns the process id of this child process

12
src/libstd/io/signal.rs
View File

@ -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()) {
let mut io = LocalIo::borrow();
match io.get().signal(signum, self.chan.clone()) {
Ok(w) => {
self.handles.insert(signum, w);
Some(())
true
},
Err(ioerr) => {
io_error::cond.raise(ioerr);
None
false
}
}
}).is_some()
}
/// Unregisters a signal. If this listener currently had a handler

19
src/libstd/io/stdio.rs
View File

@ -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.

8
src/libstd/io/timer.rs
View File

@ -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,8 +60,8 @@ 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() {
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);
@ -69,8 +69,6 @@ impl Timer {
None
}
}
})
}
/// Blocks the current task for `msecs` milliseconds.

4
src/libstd/os.rs
View File

@ -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

5
src/libstd/rt/basic.rs
View File

@ -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)
}
}

9
src/libstd/rt/borrowck.rs
View File

@ -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() {

75
src/libstd/rt/comm.rs
View File

@ -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();
let _guard = atomic();
cone.send_deferred(());
cstream.send_deferred(());
cshared.send_deferred(());
mp.send_deferred(());
})
}
}
}

5
src/libstd/rt/deque.rs
View File

@ -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),

7
src/libstd/rt/kill.rs
View File

@ -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.

86
src/libstd/rt/local.rs
View File

@ -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,10 +184,10 @@ 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 _ = Local::borrow(None::<Task>);
}
let task: ~Task = Local::take();
cleanup_task(task);
}

3
src/libstd/rt/local_heap.rs
View File

@ -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)]

47
src/libstd/rt/local_ptr.rs
View File

@ -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

33
src/libstd/rt/mod.rs
View File

@ -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);

58
src/libstd/rt/rtio.rs
View File

@ -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> {
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;
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),
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
}
}
}
// 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 {

72
src/libstd/rt/sched.rs
View File

@ -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());
});
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);
});

59
src/libstd/rt/task.rs
View File

@ -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,
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.take(),
f.take());
running_task.sched = Some(sched);
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 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.take(),
f.take());
running_task.sched = Some(sched);
home,
f);
running_task.get().sched = Some(sched);
new_task
})
}
pub fn build_root(stack_size: Option<uint>, f: proc()) -> ~Task {
@ -371,10 +365,10 @@ 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 {
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
@ -390,7 +384,6 @@ impl Task {
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,8 +588,12 @@ 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>");
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
@ -603,7 +601,6 @@ pub extern "C" fn rust_stack_exhausted() {
// 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");
}

30
src/libstd/rt/test.rs
View File

@ -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;

31
src/libstd/rt/tube.rs
View File

@ -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; }
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();
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);

17
src/libstd/run.rs
View File

@ -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() {
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() {
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();

47
src/libstd/select.rs
View File

@ -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(());
}
}
}

29
src/libstd/task/mod.rs
View File

@ -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 {
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(());

16
src/libstd/task/spawn.rs
View File

@ -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);
}

12
src/libstd/unstable/dynamic_lib.rs
View File

@ -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,11 +158,11 @@ 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(|| {
let _guard = atomic();
lock.lock();
let _old_error = dlerror();
@ -176,7 +176,6 @@ pub mod dl {
};
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,7 +225,7 @@ pub mod dl {
pub fn check_for_errors_in<T>(f: || -> T) -> Result<T, ~str> {
unsafe {
atomically(|| {
let _guard = atomic();
SetLastError(0);
let result = f();
@ -237,7 +236,6 @@ pub mod dl {
} else {
Err(format!("Error code {}", error))
}
})
}
}

4
src/libstd/unstable/mod.rs
View File

@ -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();

113
src/libstd/unstable/sync.rs
View File

@ -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();
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.take();
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()
}).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());
}
})
}
} 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())
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]

8
src/test/bench/msgsend-ring-mutex-arcs.rs
View File

@ -20,7 +20,6 @@ extern mod extra;
use extra::arc;
use extra::future::Future;
use extra::time;
use std::cell::Cell;
use std::os;
use std::uint;
@ -91,12 +90,9 @@ fn main() {
for i in range(1u, num_tasks) {
//error!("spawning %?", i);
let (new_chan, num_port) = init();
let num_chan2 = Cell::new(num_chan);
let num_port = Cell::new(num_port);
let num_chan_2 = num_chan.clone();
let new_future = do Future::spawn() {
let num_chan = num_chan2.take();
let num_port1 = num_port.take();
thread_ring(i, msg_per_task, num_chan, num_port1)
thread_ring(i, msg_per_task, num_chan_2, num_port)
};
futures.push(new_future);
num_chan = new_chan;

8
src/test/bench/msgsend-ring-rw-arcs.rs
View File

@ -20,7 +20,6 @@ extern mod extra;
use extra::arc;
use extra::future::Future;
use extra::time;
use std::cell::Cell;
use std::os;
use std::uint;
@ -87,12 +86,9 @@ fn main() {
for i in range(1u, num_tasks) {
//error!("spawning %?", i);
let (new_chan, num_port) = init();
let num_chan2 = Cell::new(num_chan);
let num_port = Cell::new(num_port);
let num_chan_2 = num_chan.clone();
let new_future = do Future::spawn {
let num_chan = num_chan2.take();
let num_port1 = num_port.take();
thread_ring(i, msg_per_task, num_chan, num_port1)
thread_ring(i, msg_per_task, num_chan_2, num_port)
};
futures.push(new_future);
num_chan = new_chan;

14
src/test/bench/rt-messaging-ping-pong.rs
View File

@ -13,7 +13,6 @@ extern mod extra;
use std::os;
use std::uint;
use std::rt::test::spawntask_later;
use std::cell::Cell;
// This is a simple bench that creates M pairs of of tasks. These
// tasks ping-pong back and forth over a pair of streams. This is a
@ -29,14 +28,9 @@ fn ping_pong_bench(n: uint, m: uint) {
// Create a stream B->A
let (pb,cb) = stream::<()>();
let pa = Cell::new(pa);
let ca = Cell::new(ca);
let pb = Cell::new(pb);
let cb = Cell::new(cb);
do spawntask_later() || {
let chan = ca.take();
let port = pb.take();
let chan = ca;
let port = pb;
n.times(|| {
chan.send(());
port.recv();
@ -44,8 +38,8 @@ fn ping_pong_bench(n: uint, m: uint) {
}
do spawntask_later() || {
let chan = cb.take();
let port = pa.take();
let chan = cb;
let port = pa;
n.times(|| {
port.recv();
chan.send(());

4
src/test/bench/rt-parfib.rs
View File

@ -13,7 +13,6 @@ extern mod extra;
use std::os;
use std::uint;
use std::rt::test::spawntask_later;
use std::cell::Cell;
use std::comm::oneshot;
// A simple implementation of parfib. One subtree is found in a new
@ -26,9 +25,8 @@ fn parfib(n: uint) -> uint {
}
let (port,chan) = oneshot::<uint>();
let chan = Cell::new(chan);
do spawntask_later {
chan.take().send(parfib(n-1));
chan.send(parfib(n-1));
};
let m2 = parfib(n-2);
return (port.recv() + m2);

9
src/test/bench/shootout-chameneos-redux.rs
View File

@ -12,7 +12,6 @@
extern mod extra;
use std::cell::Cell;
use std::comm::{stream, SharedChan};
use std::option;
use std::os;
@ -156,9 +155,11 @@ fn rendezvous(nn: uint, set: ~[color]) {
let to_rendezvous = to_rendezvous.clone();
let to_rendezvous_log = to_rendezvous_log.clone();
let (from_rendezvous, to_creature) = stream();
let from_rendezvous = Cell::new(from_rendezvous);
do task::spawn || {
creature(ii, col, from_rendezvous.take(), to_rendezvous.clone(),
do task::spawn {
creature(ii,
col,
from_rendezvous,
to_rendezvous.clone(),
to_rendezvous_log.clone());
}
to_creature

8
src/test/bench/shootout-k-nucleotide-pipes.rs
View File

@ -189,8 +189,12 @@ fn main() {
let mut proc_mode = false;
loop {
let line = match io::ignore_io_error(|| rdr.read_line()) {
Some(ln) => ln, None => break,
let line = {
let _guard = io::ignore_io_error();
match rdr.read_line() {
Some(ln) => ln,
None => break,
}
};
let line = line.trim().to_owned();

3
src/test/bench/task-perf-jargon-metal-smoke.rs
View File

@ -17,7 +17,6 @@
//
// The filename is a song reference; google it in quotes.
use std::cell::Cell;
use std::comm;
use std::os;
use std::task;
@ -27,9 +26,7 @@ fn child_generation(gens_left: uint, c: comm::Chan<()>) {
// This used to be O(n^2) in the number of generations that ever existed.
// With this code, only as many generations are alive at a time as tasks
// alive at a time,
let c = Cell::new(c);
do spawn {
let c = c.take();
if gens_left & 1 == 1 {
task::deschedule(); // shake things up a bit
}

5
src/test/compile-fail/no-send-res-ports.rs
View File

@ -10,7 +10,6 @@
#[feature(managed_boxes)];
use std::cell::Cell;
use std::task;
struct Port<T>(@T);
@ -31,10 +30,10 @@ fn main() {
}
}
let x = Cell::new(foo(Port(@())));
let x = foo(Port(@()));
do task::spawn {
let y = x.take(); //~ ERROR does not fulfill `Send`
let y = x; //~ ERROR does not fulfill `Send`
error!("{:?}", y);
}
}

6
src/test/compile-fail/no_freeze-rc.rs
View File

@ -9,11 +9,11 @@
// except according to those terms.
use std::rc::Rc;
use std::cell::Cell;
use std::cell::RefCell;
fn bar<T: Freeze>(_: T) {}
fn main() {
let x = Rc::from_send(Cell::new(5));
bar(x); //~ ERROR instantiating a type parameter with an incompatible type `std::rc::Rc<std::cell::Cell<int>>`, which does not fulfill `Freeze`
let x = Rc::from_send(RefCell::new(5));
bar(x); //~ ERROR instantiating a type parameter with an incompatible type `std::rc::Rc<std::cell::RefCell<int>>`, which does not fulfill `Freeze`
}

2
src/test/run-pass/issue-2718.rs
View File

@ -313,8 +313,6 @@ pub fn main() {
// Commented out because of option::get error
let (client_, server_) = pingpong::init();
let client_ = Cell::new(client_);
let server_ = Cell::new(server_);
task::spawn {|client_|
let client__ = client_.take();

4
src/test/run-pass/sendfn-spawn-with-fn-arg.rs
View File

@ -8,7 +8,6 @@
// option. This file may not be copied, modified, or distributed
// except according to those terms.
use std::cell::Cell;
use std::task;
pub fn main() { test05(); }
@ -23,8 +22,7 @@ fn test05() {
error!("{}", *three + n); // will copy x into the closure
assert_eq!(*three, 3);
};
let fn_to_send = Cell::new(fn_to_send);
task::spawn(proc() {
test05_start(fn_to_send.take());
test05_start(fn_to_send);
});
}

4
src/test/run-pass/task-killjoin-rsrc.rs
View File

@ -13,7 +13,6 @@
// A port of task-killjoin to use a class with a dtor to manage
// the join.
use std::cell::Cell;
use std::comm::*;
use std::ptr;
use std::task;
@ -55,9 +54,8 @@ fn joinable(f: proc()) -> Port<bool> {
*b = true;
}
let (p, c) = stream();
let c = Cell::new(c);
do task::spawn_unlinked {
let ccc = c.take();
let ccc = c;
wrapper(ccc, f)
}
p

8
src/test/run-pass/tempfile.rs
View File

@ -22,11 +22,10 @@
extern mod extra;
use extra::tempfile::TempDir;
use std::io::fs;
use std::io;
use std::os;
use std::task;
use std::cell::Cell;
use std::io;
use std::io::fs;
fn test_tempdir() {
let path = {
@ -51,9 +50,8 @@ fn test_rm_tempdir() {
let tmp = TempDir::new("test_rm_tempdir").unwrap();
let path = tmp.path().clone();
let cell = Cell::new(tmp);
let f: proc() = proc() {
let _tmp = cell.take();
let _tmp = tmp;
fail!("fail to unwind past `tmp`");
};
task::try(f);

14
src/test/run-pass/trait-bounds-in-arc.rs
View File

@ -16,10 +16,10 @@
// xfail-fast
extern mod extra;
use extra::arc;
use std::comm;
use std::task;
use std::cell;
trait Pet {
fn name(&self, blk: |&str|);
@ -71,14 +71,14 @@ fn main() {
~fishe as ~Pet:Freeze+Send,
~dogge2 as ~Pet:Freeze+Send]);
let (p1,c1) = comm::stream();
let arc1 = cell::Cell::new(arc.clone());
do task::spawn { check_legs(arc1.take()); c1.send(()); }
let arc1 = arc.clone();
do task::spawn { check_legs(arc1); c1.send(()); }
let (p2,c2) = comm::stream();
let arc2 = cell::Cell::new(arc.clone());
do task::spawn { check_names(arc2.take()); c2.send(()); }
let arc2 = arc.clone();
do task::spawn { check_names(arc2); c2.send(()); }
let (p3,c3) = comm::stream();
let arc3 = cell::Cell::new(arc.clone());
do task::spawn { check_pedigree(arc3.take()); c3.send(()); }
let arc3 = arc.clone();
do task::spawn { check_pedigree(arc3); c3.send(()); }
p1.recv();
p2.recv();
p3.recv();