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

Author: Eric Reed

src/libstd/macros.rs

@@ -10,18 +10,16 @@
 
 #[macro_escape];
 
+macro_rules! rterrln (
+    ($( $arg:expr),+) => ( {
+        ::rt::util::dumb_println(fmt!( $($arg),+ ));
+    } )
+)
+
 // Some basic logging
 macro_rules! rtdebug_ (
     ($( $arg:expr),+) => ( {
-        dumb_println(fmt!( $($arg),+ ));
-
-        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");
-        }
-
+        rterrln!( $($arg),+ )
     } )
 )
 
@@ -33,24 +31,15 @@ macro_rules! rtdebug (
 macro_rules! rtassert (
     ( $arg:expr ) => ( {
         if !$arg {
-            abort!("assertion failed: %s", stringify!($arg));
+            rtabort!("assertion failed: %s", stringify!($arg));
         }
     } )
 )
 
 
-// The do_abort function was originally inside the abort macro, but
-// this was ICEing the compiler so it has been moved outside. Now this
-// seems to work?
-#[allow(missing_doc)]
-pub fn do_abort() -> ! {
-    unsafe { ::libc::abort(); }
-}
-
-macro_rules! abort(
+macro_rules! rtabort(
     ($( $msg:expr),+) => ( {
-        rtdebug!($($msg),+);
-        ::macros::do_abort();
+        ::rt::util::abort(fmt!($($msg),+));
     } )
 )
 

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?

src/libstd/rt/local.rs

@@ -38,25 +38,25 @@ 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 {
                 Some(~ref mut 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)]

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

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