libs: Fix miscellaneous fallout of librustrt

6 files changed, 55 insertions, 64 deletions

diff --git a/src/libgreen/basic.rs b/src/libgreen/basic.rs
index 1ebebbe555e..7f033a1bc61 100644
--- a/src/libgreen/basic.rs
+++ b/src/libgreen/basic.rs
@@ -20,7 +20,7 @@ use std::sync::atomics;
 use std::mem;
 use std::rt::rtio::{EventLoop, IoFactory, RemoteCallback};
 use std::rt::rtio::{PausableIdleCallback, Callback};
-use std::unstable::sync::Exclusive;
+use std::rt::exclusive::Exclusive;
 
 /// This is the only exported function from this module.
 pub fn event_loop() -> Box<EventLoop:Send> {
@@ -31,7 +31,7 @@ struct BasicLoop {
     work: Vec<proc():Send>,             // pending work
     remotes: Vec<(uint, Box<Callback:Send>)>,
     next_remote: uint,
-    messages: Exclusive<Vec<Message>>,
+    messages: Arc<Exclusive<Vec<Message>>>,
     idle: Option<Box<Callback:Send>>,
     idle_active: Option<Arc<atomics::AtomicBool>>,
 }
@@ -46,7 +46,7 @@ impl BasicLoop {
             idle_active: None,
             next_remote: 0,
             remotes: vec![],
-            messages: Exclusive::new(vec![]),
+            messages: Arc::new(Exclusive::new(Vec::new())),
         }
     }
 
@@ -61,19 +61,10 @@ impl BasicLoop {
 
     fn remote_work(&mut self) {
         let messages = unsafe {
-            self.messages.with(|messages| {
-                if messages.len() > 0 {
-                    Some(mem::replace(messages, vec![]))
-                } else {
-                    None
-                }
-            })
-        };
-        let messages = match messages {
-            Some(m) => m, None => return
+            mem::replace(&mut *self.messages.lock(), Vec::new())
         };
-        for message in messages.iter() {
-            self.message(*message);
+        for message in messages.move_iter() {
+            self.message(message);
         }
     }
 
@@ -125,13 +116,13 @@ impl EventLoop for BasicLoop {
             }
 
             unsafe {
+                let mut messages = self.messages.lock();
                 // We block here if we have no messages to process and we may
                 // receive a message at a later date
-                self.messages.hold_and_wait(|messages| {
-                    self.remotes.len() > 0 &&
-                        messages.len() == 0 &&
-                        self.work.len() == 0
-                })
+                if self.remotes.len() > 0 && messages.len() == 0 &&
+                   self.work.len() == 0 {
+                    messages.wait()
+                }
             }
         }
     }
@@ -165,33 +156,29 @@ impl EventLoop for BasicLoop {
 }
 
 struct BasicRemote {
-    queue: Exclusive<Vec<Message>>,
+    queue: Arc<Exclusive<Vec<Message>>>,
     id: uint,
 }
 
 impl BasicRemote {
-    fn new(queue: Exclusive<Vec<Message>>, id: uint) -> BasicRemote {
+    fn new(queue: Arc<Exclusive<Vec<Message>>>, id: uint) -> BasicRemote {
         BasicRemote { queue: queue, id: id }
     }
 }
 
 impl RemoteCallback for BasicRemote {
     fn fire(&mut self) {
-        unsafe {
-            self.queue.hold_and_signal(|queue| {
-                queue.push(RunRemote(self.id));
-            })
-        }
+        let mut queue = unsafe { self.queue.lock() };
+        queue.push(RunRemote(self.id));
+        queue.signal();
     }
 }
 
 impl Drop for BasicRemote {
     fn drop(&mut self) {
-        unsafe {
-            self.queue.hold_and_signal(|queue| {
-                queue.push(RemoveRemote(self.id));
-            })
-        }
+        let mut queue = unsafe { self.queue.lock() };
+        queue.push(RemoveRemote(self.id));
+        queue.signal();
     }
 }
 
@@ -216,7 +203,7 @@ impl Drop for BasicPausable {
 
 #[cfg(test)]
 mod test {
-    use std::task::TaskOpts;
+    use std::rt::task::TaskOpts;
 
     use basic;
     use PoolConfig;
diff --git a/src/libgreen/lib.rs b/src/libgreen/lib.rs
index c75d69480ce..333ac80907f 100644
--- a/src/libgreen/lib.rs
+++ b/src/libgreen/lib.rs
@@ -160,7 +160,7 @@
 //! # Using a scheduler pool
 //!
 //! ```rust
-//! use std::task::TaskOpts;
+//! use std::rt::task::TaskOpts;
 //! use green::{SchedPool, PoolConfig};
 //! use green::sched::{PinnedTask, TaskFromFriend};
 //!
@@ -221,10 +221,10 @@ use std::mem::replace;
 use std::os;
 use std::rt::rtio;
 use std::rt::thread::Thread;
+use std::rt::task::TaskOpts;
 use std::rt;
 use std::sync::atomics::{SeqCst, AtomicUint, INIT_ATOMIC_UINT};
 use std::sync::deque;
-use std::task::TaskOpts;
 
 use sched::{Shutdown, Scheduler, SchedHandle, TaskFromFriend, NewNeighbor};
 use sleeper_list::SleeperList;
@@ -319,7 +319,7 @@ pub fn run(event_loop_factory: fn() -> Box<rtio::EventLoop:Send>,
     let mut pool = SchedPool::new(cfg);
     let (tx, rx) = channel();
     let mut opts = TaskOpts::new();
-    opts.notify_chan = Some(tx);
+    opts.on_exit = Some(proc(r) tx.send(r));
     opts.name = Some("<main>".into_maybe_owned());
     pool.spawn(opts, main);
 
diff --git a/src/libgreen/sched.rs b/src/libgreen/sched.rs
index e410d2719b1..f55dc92eac6 100644
--- a/src/libgreen/sched.rs
+++ b/src/libgreen/sched.rs
@@ -10,11 +10,11 @@
 
 use std::mem;
 use std::rt::local::Local;
+use std::rt::mutex::NativeMutex;
 use std::rt::rtio::{RemoteCallback, PausableIdleCallback, Callback, EventLoop};
 use std::rt::task::BlockedTask;
 use std::rt::task::Task;
 use std::sync::deque;
-use std::unstable::mutex::NativeMutex;
 use std::raw;
 
 use std::rand::{XorShiftRng, Rng, Rand};
@@ -1022,7 +1022,7 @@ fn new_sched_rng() -> XorShiftRng {
 mod test {
     use rustuv;
 
-    use std::task::TaskOpts;
+    use std::rt::task::TaskOpts;
     use std::rt::task::Task;
     use std::rt::local::Local;
 
@@ -1475,7 +1475,7 @@ mod test {
 
     #[test]
     fn test_spawn_sched_blocking() {
-        use std::unstable::mutex::{StaticNativeMutex, NATIVE_MUTEX_INIT};
+        use std::rt::mutex::{StaticNativeMutex, NATIVE_MUTEX_INIT};
         static mut LOCK: StaticNativeMutex = NATIVE_MUTEX_INIT;
 
         // Testing that a task in one scheduler can block in foreign code
diff --git a/src/libgreen/simple.rs b/src/libgreen/simple.rs
index 49aef15f93b..7b738ed9c7c 100644
--- a/src/libgreen/simple.rs
+++ b/src/libgreen/simple.rs
@@ -15,10 +15,9 @@ use std::any::Any;
 use std::mem;
 use std::rt::Runtime;
 use std::rt::local::Local;
+use std::rt::mutex::NativeMutex;
 use std::rt::rtio;
-use std::rt::task::{Task, BlockedTask};
-use std::task::TaskOpts;
-use std::unstable::mutex::NativeMutex;
+use std::rt::task::{Task, BlockedTask, TaskOpts};
 
 struct SimpleTask {
     lock: NativeMutex,
diff --git a/src/libgreen/stack.rs b/src/libgreen/stack.rs
index f42d636cafb..2e385f75e1d 100644
--- a/src/libgreen/stack.rs
+++ b/src/libgreen/stack.rs
@@ -8,9 +8,9 @@
 // option. This file may not be copied, modified, or distributed
 // except according to those terms.
 
-use std::rt::env::max_cached_stacks;
+use std::sync::atomics;
 use std::os::{errno, page_size, MemoryMap, MapReadable, MapWritable,
-              MapNonStandardFlags, MapVirtual};
+              MapNonStandardFlags, MapVirtual, getenv};
 use libc;
 
 /// A task's stack. The name "Stack" is a vestige of segmented stacks.
@@ -151,6 +151,22 @@ impl StackPool {
     }
 }
 
+fn max_cached_stacks() -> uint {
+    static mut AMT: atomics::AtomicUint = atomics::INIT_ATOMIC_UINT;
+    match unsafe { AMT.load(atomics::SeqCst) } {
+        0 => {}
+        n => return n - 1,
+    }
+    let amt = getenv("RUST_MAX_CACHED_STACKS").and_then(|s| from_str(s.as_slice()));
+    // This default corresponds to 20M of cache per scheduler (at the
+    // default size).
+    let amt = amt.unwrap_or(10);
+    // 0 is our sentinel value, so ensure that we'll never see 0 after
+    // initialization has run
+    unsafe { AMT.store(amt + 1, atomics::SeqCst); }
+    return amt;
+}
+
 extern {
     fn rust_valgrind_stack_register(start: *libc::uintptr_t,
                                     end: *libc::uintptr_t) -> libc::c_uint;
diff --git a/src/libgreen/task.rs b/src/libgreen/task.rs
index f50f65af6f9..91ebad3b3f8 100644
--- a/src/libgreen/task.rs
+++ b/src/libgreen/task.rs
@@ -22,13 +22,12 @@ use std::any::Any;
 use std::mem;
 use std::raw;
 use std::rt::Runtime;
-use std::rt::env;
 use std::rt::local::Local;
+use std::rt::mutex::NativeMutex;
 use std::rt::rtio;
 use std::rt::stack;
-use std::rt::task::{Task, BlockedTask, SendMessage};
-use std::task::TaskOpts;
-use std::unstable::mutex::NativeMutex;
+use std::rt::task::{Task, BlockedTask, TaskOpts};
+use std::rt;
 
 use context::Context;
 use coroutine::Coroutine;
@@ -142,7 +141,7 @@ impl GreenTask {
         let mut ops = GreenTask::new_typed(None, TypeGreen(Some(home)));
 
         // Allocate a stack for us to run on
-        let stack_size = stack_size.unwrap_or_else(|| env::min_stack());
+        let stack_size = stack_size.unwrap_or_else(|| rt::min_stack());
         let mut stack = stack_pool.take_stack(stack_size);
         let context = Context::new(bootstrap_green_task, ops.as_uint(), start,
                                    &mut stack);
@@ -176,23 +175,13 @@ impl GreenTask {
     pub fn configure(pool: &mut StackPool,
                      opts: TaskOpts,
                      f: proc():Send) -> Box<GreenTask> {
-        let TaskOpts {
-            notify_chan, name, stack_size,
-            stderr, stdout,
-        } = opts;
+        let TaskOpts { name, stack_size, on_exit } = opts;
 
         let mut green = GreenTask::new(pool, stack_size, f);
         {
             let task = green.task.get_mut_ref();
             task.name = name;
-            task.stderr = stderr;
-            task.stdout = stdout;
-            match notify_chan {
-                Some(chan) => {
-                    task.death.on_exit = Some(SendMessage(chan));
-                }
-                None => {}
-            }
+            task.death.on_exit = on_exit;
         }
         return green;
     }
@@ -490,7 +479,7 @@ mod tests {
     use std::rt::local::Local;
     use std::rt::task::Task;
     use std::task;
-    use std::task::TaskOpts;
+    use std::rt::task::TaskOpts;
 
     use super::super::{PoolConfig, SchedPool};
     use super::GreenTask;
@@ -529,7 +518,7 @@ mod tests {
         opts.name = Some("test".into_maybe_owned());
         opts.stack_size = Some(20 * 4096);
         let (tx, rx) = channel();
-        opts.notify_chan = Some(tx);
+        opts.on_exit = Some(proc(r) tx.send(r));
         spawn_opts(opts, proc() {});
         assert!(rx.recv().is_ok());
     }
@@ -538,7 +527,7 @@ mod tests {
     fn smoke_opts_fail() {
         let mut opts = TaskOpts::new();
         let (tx, rx) = channel();
-        opts.notify_chan = Some(tx);
+        opts.on_exit = Some(proc(r) tx.send(r));
         spawn_opts(opts, proc() { fail!() });
         assert!(rx.recv().is_err());
     }