Merge remote-tracking branch 'upstream/io' into io

7 files changed, 239 insertions, 62 deletions

diff --git a/src/libstd/rt/global_heap.rs b/src/libstd/rt/global_heap.rs
index ce7ff87b445..e89df2b1c93 100644
--- a/src/libstd/rt/global_heap.rs
+++ b/src/libstd/rt/global_heap.rs
@@ -14,7 +14,7 @@ use c_malloc = libc::malloc;
 use c_free = libc::free;
 use managed::raw::{BoxHeaderRepr, BoxRepr};
 use cast::transmute;
-use unstable::intrinsics::{atomic_xadd,atomic_xsub};
+use unstable::intrinsics::{atomic_xadd,atomic_xsub, atomic_load};
 use ptr::null;
 use intrinsic::TyDesc;
 
@@ -34,8 +34,7 @@ pub unsafe fn malloc(td: *TypeDesc, size: uint) -> *c_void {
     box.header.prev = null();
     box.header.next = null();
 
-    let exchange_count = &mut *exchange_count_ptr();
-    atomic_xadd(exchange_count, 1);
+    inc_count();
 
     return transmute(box);
 }
@@ -48,21 +47,46 @@ pub unsafe fn malloc_raw(size: uint) -> *c_void {
     if p.is_null() {
         fail!("Failure in malloc_raw: result ptr is null");
     }
+    inc_count();
     p
 }
 
 pub unsafe fn free(ptr: *c_void) {
-    let exchange_count = &mut *exchange_count_ptr();
-    atomic_xsub(exchange_count, 1);
-
     assert!(ptr.is_not_null());
+    dec_count();
     c_free(ptr);
 }
 ///Thin wrapper around libc::free, as with exchange_alloc::malloc_raw
 pub unsafe fn free_raw(ptr: *c_void) {
+    assert!(ptr.is_not_null());
+    dec_count();
     c_free(ptr);
 }
 
+fn inc_count() {
+    unsafe {
+        let exchange_count = &mut *exchange_count_ptr();
+        atomic_xadd(exchange_count, 1);
+    }
+}
+
+fn dec_count() {
+    unsafe {
+        let exchange_count = &mut *exchange_count_ptr();
+        atomic_xsub(exchange_count, 1);
+    }
+}
+
+pub fn cleanup() {
+    unsafe {
+        let count_ptr = exchange_count_ptr();
+        let allocations = atomic_load(&*count_ptr);
+        if allocations != 0 {
+            rtabort!("exchange heap not empty on exit - %i dangling allocations", allocations);
+        }
+    }
+}
+
 fn get_box_size(body_size: uint, body_align: uint) -> uint {
     let header_size = size_of::<BoxHeaderRepr>();
     // FIXME (#2699): This alignment calculation is suspicious. Is it right?
diff --git a/src/libstd/rt/local.rs b/src/libstd/rt/local.rs
index 6e0fbda5ec9..6df1ffaa453 100644
--- a/src/libstd/rt/local.rs
+++ b/src/libstd/rt/local.rs
@@ -38,17 +38,17 @@ impl Local for Scheduler {
         }
         match res {
             Some(r) => { r }
-            None => abort!("function failed!")
+            None => rtabort!("function failed!")
         }
     }
     unsafe fn unsafe_borrow() -> *mut Scheduler { local_ptr::unsafe_borrow() }
-    unsafe fn try_unsafe_borrow() -> Option<*mut Scheduler> { abort!("unimpl") }
+    unsafe fn try_unsafe_borrow() -> Option<*mut Scheduler> { rtabort!("unimpl") }
 }
 
 impl Local for Task {
-    fn put(_value: ~Task) { abort!("unimpl") }
-    fn take() -> ~Task { abort!("unimpl") }
-    fn exists() -> bool { abort!("unimpl") }
+    fn put(_value: ~Task) { rtabort!("unimpl") }
+    fn take() -> ~Task { rtabort!("unimpl") }
+    fn exists() -> bool { rtabort!("unimpl") }
     fn borrow<T>(f: &fn(&mut Task) -> T) -> T {
         do Local::borrow::<Scheduler, T> |sched| {
             match sched.current_task {
@@ -56,7 +56,7 @@ impl Local for Task {
                     f(&mut *task.task)
                 }
                 None => {
-                    abort!("no scheduler")
+                    rtabort!("no scheduler")
                 }
             }
         }
@@ -69,7 +69,7 @@ impl Local for Task {
             }
             None => {
                 // Don't fail. Infinite recursion
-                abort!("no scheduler")
+                rtabort!("no scheduler")
             }
         }
     }
@@ -84,16 +84,16 @@ impl Local for Task {
 
 // XXX: This formulation won't work once ~IoFactoryObject is a real trait pointer
 impl Local for IoFactoryObject {
-    fn put(_value: ~IoFactoryObject) { abort!("unimpl") }
-    fn take() -> ~IoFactoryObject { abort!("unimpl") }
-    fn exists() -> bool { abort!("unimpl") }
-    fn borrow<T>(_f: &fn(&mut IoFactoryObject) -> T) -> T { abort!("unimpl") }
+    fn put(_value: ~IoFactoryObject) { rtabort!("unimpl") }
+    fn take() -> ~IoFactoryObject { rtabort!("unimpl") }
+    fn exists() -> bool { rtabort!("unimpl") }
+    fn borrow<T>(_f: &fn(&mut IoFactoryObject) -> T) -> T { rtabort!("unimpl") }
     unsafe fn unsafe_borrow() -> *mut IoFactoryObject {
         let sched = Local::unsafe_borrow::<Scheduler>();
         let io: *mut IoFactoryObject = (*sched).event_loop.io().unwrap();
         return io;
     }
-    unsafe fn try_unsafe_borrow() -> Option<*mut IoFactoryObject> { abort!("unimpl") }
+    unsafe fn try_unsafe_borrow() -> Option<*mut IoFactoryObject> { rtabort!("unimpl") }
 }
 
 #[cfg(test)]
diff --git a/src/libstd/rt/local_ptr.rs b/src/libstd/rt/local_ptr.rs
index 0db903f81ee..cd7c5daa444 100644
--- a/src/libstd/rt/local_ptr.rs
+++ b/src/libstd/rt/local_ptr.rs
@@ -109,7 +109,7 @@ pub unsafe fn unsafe_borrow<T>() -> *mut T {
 fn tls_key() -> tls::Key {
     match maybe_tls_key() {
         Some(key) => key,
-        None => abort!("runtime tls key not initialized")
+        None => rtabort!("runtime tls key not initialized")
     }
 }
 
diff --git a/src/libstd/rt/mod.rs b/src/libstd/rt/mod.rs
index 1724361cabc..581e3addff0 100644
--- a/src/libstd/rt/mod.rs
+++ b/src/libstd/rt/mod.rs
@@ -60,7 +60,21 @@ Several modules in `core` are clients of `rt`:
 #[deny(unused_variable)];
 
 use cell::Cell;
+use clone::Clone;
+use container::Container;
+use from_str::FromStr;
+use iterator::IteratorUtil;
+use option::{Some, None};
+use os;
 use ptr::RawPtr;
+use uint;
+use rt::sched::{Scheduler, Coroutine, Shutdown};
+use rt::sleeper_list::SleeperList;
+use rt::task::Task;
+use rt::thread::Thread;
+use rt::work_queue::WorkQueue;
+use rt::uv::uvio::UvEventLoop;
+use vec::{OwnedVector, MutableVector};
 
 /// The global (exchange) heap.
 pub mod global_heap;
@@ -139,6 +153,9 @@ pub mod join_latch;
 
 pub mod metrics;
 
+// FIXME #5248 shouldn't be pub
+/// Just stuff
+pub mod util;
 
 /// Set up a default runtime configuration, given compiler-supplied arguments.
 ///
@@ -156,22 +173,9 @@ pub mod metrics;
 /// The return value is used as the process return code. 0 on success, 101 on error.
 pub fn start(_argc: int, _argv: **u8, crate_map: *u8, main: ~fn()) -> int {
 
-    use self::sched::{Scheduler, Coroutine};
-    use self::work_queue::WorkQueue;
-    use self::uv::uvio::UvEventLoop;
-    use self::sleeper_list::SleeperList;
-
     init(crate_map);
-
-    let loop_ = ~UvEventLoop::new();
-    let work_queue = WorkQueue::new();
-    let sleepers = SleeperList::new();
-    let mut sched = ~Scheduler::new(loop_, work_queue, sleepers);
-    sched.no_sleep = true;
-    let main_task = ~Coroutine::new_root(&mut sched.stack_pool, main);
-
-    sched.enqueue_task(main_task);
-    sched.run();
+    run(main);
+    cleanup();
 
     return 0;
 }
@@ -182,6 +186,71 @@ pub fn init(crate_map: *u8) {
     logging::init(crate_map);
 }
 
+pub fn cleanup() {
+    global_heap::cleanup();
+}
+
+pub fn run(main: ~fn()) {
+    let nthreads = match os::getenv("RUST_THREADS") {
+        Some(nstr) => FromStr::from_str(nstr).get(),
+        None => unsafe {
+            // Using more threads than cores in test code
+            // to force the OS to preempt them frequently.
+            // Assuming that this help stress test concurrent types.
+            util::num_cpus() * 2
+        }
+    };
+
+    let sleepers = SleeperList::new();
+    let work_queue = WorkQueue::new();
+
+    let mut handles = ~[];
+    let mut scheds = ~[];
+
+    for uint::range(0, nthreads) |_| {
+        let loop_ = ~UvEventLoop::new();
+        let mut sched = ~Scheduler::new(loop_, work_queue.clone(), sleepers.clone());
+        let handle = sched.make_handle();
+
+        handles.push(handle);
+        scheds.push(sched);
+    }
+
+    let main_cell = Cell::new(main);
+    let handles = Cell::new(handles);
+    let mut new_task = ~Task::new_root();
+    let on_exit: ~fn(bool) = |exit_status| {
+
+        let mut handles = handles.take();
+        // Tell schedulers to exit
+        for handles.mut_iter().advance |handle| {
+            handle.send(Shutdown);
+        }
+
+        rtassert!(exit_status);
+    };
+    new_task.on_exit = Some(on_exit);
+    let main_task = ~Coroutine::with_task(&mut scheds[0].stack_pool,
+                                          new_task, main_cell.take());
+    scheds[0].enqueue_task(main_task);
+
+    let mut threads = ~[];
+
+    while !scheds.is_empty() {
+        let sched = scheds.pop();
+        let sched_cell = Cell::new(sched);
+        let thread = do Thread::start {
+            let sched = sched_cell.take();
+            sched.run();
+        };
+
+        threads.push(thread);
+    }
+
+    // Wait for schedulers
+    let _threads = threads;
+}
+
 /// Possible contexts in which Rust code may be executing.
 /// Different runtime services are available depending on context.
 /// Mostly used for determining if we're using the new scheduler
diff --git a/src/libstd/rt/sched.rs b/src/libstd/rt/sched.rs
index be57247d514..453d3303db6 100644
--- a/src/libstd/rt/sched.rs
+++ b/src/libstd/rt/sched.rs
@@ -357,7 +357,7 @@ impl Scheduler {
                 home_handle.send(PinnedTask(task));
             }
             AnySched => {
-                abort!("error: cannot send anysched task home");
+                rtabort!("error: cannot send anysched task home");
             }
         }
     }
@@ -381,51 +381,43 @@ impl Scheduler {
                     match home {
                         &Some(Sched(ref home_handle))
                         if home_handle.sched_id != this.sched_id() => {
-                            0
+                            SendHome
                         }
                         &Some(AnySched) if this.run_anything => {
-                            1
+                            ResumeNow
                         }
                         &Some(AnySched) => {
-                            2
+                            Requeue
                         }
                         &Some(Sched(_)) => {
-                            3
+                            ResumeNow
                         }
                         &None => {
-                            4
+                            Homeless
                         }
                     }
                 };
 
                 match action_id {
-                    0 => {
+                    SendHome => {
                         rtdebug!("sending task home");
                         Scheduler::send_task_home(task);
                         Local::put(this);
                         return false;
                     }
-                    1 => {
+                    ResumeNow => {
                         rtdebug!("resuming now");
                         this.resume_task_immediately(task);
                         return true;
                     }
-                    2 => {
+                    Requeue => {
                         rtdebug!("re-queueing")
                         this.enqueue_task(task);
                         Local::put(this);
                         return false;
                     }
-                    3 => {
-                        rtdebug!("resuming now");
-                        this.resume_task_immediately(task);
-                        return true;
-                    }
-                    4 => {
-                        abort!("task home was None!");
-                    }
-                    _ => {
-                        abort!("literally, you should not be here");
+                    Homeless => {
+                        rtabort!("task home was None!");
                     }
                 }
             }
@@ -452,7 +444,7 @@ impl Scheduler {
             dead_task.take().recycle(&mut sched.stack_pool);
         }
 
-        abort!("control reached end of task");
+        rtabort!("control reached end of task");
     }
 
     pub fn schedule_task(~self, task: ~Coroutine) {
@@ -654,6 +646,14 @@ impl Scheduler {
     }
 }
 
+// The cases for the below function.                                              
+enum ResumeAction {
+    SendHome,
+    Requeue,
+    ResumeNow,
+    Homeless                                                                      
+}
+
 impl SchedHandle {
     pub fn send(&mut self, msg: SchedMessage) {
         self.queue.push(msg);
@@ -672,7 +672,7 @@ impl Coroutine {
                 Some(Sched(SchedHandle { sched_id: ref id, _ })) => {
                     *id == sched.sched_id()
                 }
-                None => { abort!("error: homeless task!"); }
+                None => { rtabort!("error: homeless task!"); }
             }
         }
     }
@@ -696,7 +696,7 @@ impl Coroutine {
         match self.task.home {
             Some(AnySched) => { false }
             Some(Sched(_)) => { true }
-            None => { abort!("error: homeless task!");
+            None => { rtabort!("error: homeless task!");
                     }
         }
     }
@@ -710,7 +710,7 @@ impl Coroutine {
             Some(Sched(SchedHandle { sched_id: ref id, _})) => {
                 *id == sched.sched_id()
             }
-            None => { abort!("error: homeless task!"); }
+            None => { rtabort!("error: homeless task!"); }
         }
     }
 
diff --git a/src/libstd/rt/test.rs b/src/libstd/rt/test.rs
index 6e4fb9b1d94..36efcd91834 100644
--- a/src/libstd/rt/test.rs
+++ b/src/libstd/rt/test.rs
@@ -63,11 +63,11 @@ pub fn run_in_newsched_task(f: ~fn()) {
 /// in one of the schedulers. The schedulers will stay alive
 /// until the function `f` returns.
 pub fn run_in_mt_newsched_task(f: ~fn()) {
-    use libc;
     use os;
     use from_str::FromStr;
     use rt::uv::uvio::UvEventLoop;
     use rt::sched::Shutdown;
+    use rt::util;
 
     let f_cell = Cell::new(f);
 
@@ -78,7 +78,7 @@ pub fn run_in_mt_newsched_task(f: ~fn()) {
                 // Using more threads than cores in test code
                 // to force the OS to preempt them frequently.
                 // Assuming that this help stress test concurrent types.
-                rust_get_num_cpus() * 2
+                util::num_cpus() * 2
             }
         };
 
@@ -132,9 +132,6 @@ pub fn run_in_mt_newsched_task(f: ~fn()) {
         let _threads = threads;
     }
 
-    extern {
-        fn rust_get_num_cpus() -> libc::uintptr_t;
-    }
 }
 
 // THIS IS AWFUL. Copy-pasted the above initialization function but
diff --git a/src/libstd/rt/util.rs b/src/libstd/rt/util.rs
new file mode 100644
index 00000000000..904b2f8bbb9
--- /dev/null
+++ b/src/libstd/rt/util.rs
@@ -0,0 +1,87 @@
+// Copyright 2013 The Rust Project Developers. See the COPYRIGHT
+// file at the top-level directory of this distribution and at
+// http://rust-lang.org/COPYRIGHT.
+//
+// Licensed under the Apache License, Version 2.0 <LICENSE-APACHE or
+// http://www.apache.org/licenses/LICENSE-2.0> or the MIT license
+// <LICENSE-MIT or http://opensource.org/licenses/MIT>, at your
+// option. This file may not be copied, modified, or distributed
+// except according to those terms.
+
+use container::Container;
+use iterator::IteratorUtil;
+use libc;
+use str::StrSlice;
+
+/// Get the number of cores available
+pub fn num_cpus() -> uint {
+    unsafe {
+        return rust_get_num_cpus();
+    }
+
+    extern {
+        fn rust_get_num_cpus() -> libc::uintptr_t;
+    }
+}
+
+pub fn dumb_println(s: &str) {
+    use io::WriterUtil;
+    let dbg = ::libc::STDERR_FILENO as ::io::fd_t;
+    dbg.write_str(s);
+    dbg.write_str("\n");
+}
+
+pub fn abort(msg: &str) -> ! {
+    let msg = if !msg.is_empty() { msg } else { "aborted" };
+    let hash = msg.iter().fold(0, |accum, val| accum + (val as uint) );
+    let quote = match hash % 10 {
+        0 => "
+It was from the artists and poets that the pertinent answers came, and I
+know that panic would have broken loose had they been able to compare notes.
+As it was, lacking their original letters, I half suspected the compiler of
+having asked leading questions, or of having edited the correspondence in
+corroboration of what he had latently resolved to see.",
+        1 => "
+There are not many persons who know what wonders are opened to them in the
+stories and visions of their youth; for when as children we listen and dream,
+we think but half-formed thoughts, and when as men we try to remember, we are
+dulled and prosaic with the poison of life. But some of us awake in the night
+with strange phantasms of enchanted hills and gardens, of fountains that sing
+in the sun, of golden cliffs overhanging murmuring seas, of plains that stretch
+down to sleeping cities of bronze and stone, and of shadowy companies of heroes
+that ride caparisoned white horses along the edges of thick forests; and then
+we know that we have looked back through the ivory gates into that world of
+wonder which was ours before we were wise and unhappy.",
+        2 => "
+Instead of the poems I had hoped for, there came only a shuddering blackness
+and ineffable loneliness; and I saw at last a fearful truth which no one had
+ever dared to breathe before — the unwhisperable secret of secrets — The fact
+that this city of stone and stridor is not a sentient perpetuation of Old New
+York as London is of Old London and Paris of Old Paris, but that it is in fact
+quite dead, its sprawling body imperfectly embalmed and infested with queer
+animate things which have nothing to do with it as it was in life.",
+        3 => "
+The ocean ate the last of the land and poured into the smoking gulf, thereby
+giving up all it had ever conquered. From the new-flooded lands it flowed
+again, uncovering death and decay; and from its ancient and immemorial bed it
+trickled loathsomely, uncovering nighted secrets of the years when Time was
+young and the gods unborn. Above the waves rose weedy remembered spires. The
+moon laid pale lilies of light on dead London, and Paris stood up from its damp
+grave to be sanctified with star-dust. Then rose spires and monoliths that were
+weedy but not remembered; terrible spires and monoliths of lands that men never
+knew were lands...",
+        4 => "
+There was a night when winds from unknown spaces whirled us irresistibly into
+limitless vacum beyond all thought and entity. Perceptions of the most
+maddeningly untransmissible sort thronged upon us; perceptions of infinity
+which at the time convulsed us with joy, yet which are now partly lost to my
+memory and partly incapable of presentation to others.",
+        _ => "You've met with a terrible fate, haven't you?"
+    };
+    rterrln!("%s", "");
+    rterrln!("%s", quote);
+    rterrln!("%s", "");
+    rterrln!("fatal runtime error: %s", msg);
+
+    unsafe { libc::abort(); }
+}