std::rt: Change Thread interface to require an explicit join

Makes it more obvious what's going on

8 files changed, 54 insertions, 25 deletions

diff --git a/src/libstd/rt/comm.rs b/src/libstd/rt/comm.rs
index c0effdaa94c..5bc520b8a5f 100644
--- a/src/libstd/rt/comm.rs
+++ b/src/libstd/rt/comm.rs
@@ -674,10 +674,11 @@ mod test {
             do run_in_newsched_task {
                 let (port, chan) = oneshot::<int>();
                 let port_cell = Cell::new(port);
-                let _thread = do spawntask_thread {
+                let thread = do spawntask_thread {
                     let _p = port_cell.take();
                 };
                 let _chan = chan;
+                thread.join();
             }
         }
     }
@@ -689,13 +690,15 @@ mod test {
                 let (port, chan) = oneshot::<int>();
                 let chan_cell = Cell::new(chan);
                 let port_cell = Cell::new(port);
-                let _thread1 = do spawntask_thread {
+                let thread1 = do spawntask_thread {
                     let _p = port_cell.take();
                 };
-                let _thread2 = do spawntask_thread {
+                let thread2 = do spawntask_thread {
                     let c = chan_cell.take();
                     c.send(1);
                 };
+                thread1.join();
+                thread2.join();
             }
         }
     }
@@ -707,19 +710,21 @@ mod test {
                 let (port, chan) = oneshot::<int>();
                 let chan_cell = Cell::new(chan);
                 let port_cell = Cell::new(port);
-                let _thread1 = do spawntask_thread {
+                let thread1 = do spawntask_thread {
                     let port_cell = Cell::new(port_cell.take());
                     let res = do spawntask_try {
                         port_cell.take().recv();
                     };
                     assert!(res.is_err());
                 };
-                let _thread2 = do spawntask_thread {
+                let thread2 = do spawntask_thread {
                     let chan_cell = Cell::new(chan_cell.take());
                     do spawntask {
                         chan_cell.take();
                     }
                 };
+                thread1.join();
+                thread2.join();
             }
         }
     }
@@ -731,12 +736,14 @@ mod test {
                 let (port, chan) = oneshot::<~int>();
                 let chan_cell = Cell::new(chan);
                 let port_cell = Cell::new(port);
-                let _thread1 = do spawntask_thread {
+                let thread1 = do spawntask_thread {
                     chan_cell.take().send(~10);
                 };
-                let _thread2 = do spawntask_thread {
+                let thread2 = do spawntask_thread {
                     assert!(port_cell.take().recv() == ~10);
                 };
+                thread1.join();
+                thread2.join();
             }
         }
     }
diff --git a/src/libstd/rt/mod.rs b/src/libstd/rt/mod.rs
index 808d07ce77d..e9e4b655dd7 100644
--- a/src/libstd/rt/mod.rs
+++ b/src/libstd/rt/mod.rs
@@ -343,7 +343,9 @@ fn run_(main: ~fn(), use_main_sched: bool) -> int {
     }
 
     // Wait for schedulers
-    { let _threads = threads; }
+    for threads.consume_iter().advance() |thread| {
+        thread.join();
+    }
 
     // Return the exit code
     unsafe {
diff --git a/src/libstd/rt/sched.rs b/src/libstd/rt/sched.rs
index 33cfd69fcd2..98df38f9b1d 100644
--- a/src/libstd/rt/sched.rs
+++ b/src/libstd/rt/sched.rs
@@ -901,10 +901,8 @@ mod test {
                 sched.run();
             };
 
-            // wait for the end
-            let _thread1 = normal_thread;
-            let _thread2 = special_thread;
-
+            normal_thread.join();
+            special_thread.join();
         }
     }
 
@@ -1074,16 +1072,19 @@ mod test {
             sched2.enqueue_task(task2);
 
             let sched1_cell = Cell::new(sched1);
-            let _thread1 = do Thread::start {
+            let thread1 = do Thread::start {
                 let sched1 = sched1_cell.take();
                 sched1.run();
             };
 
             let sched2_cell = Cell::new(sched2);
-            let _thread2 = do Thread::start {
+            let thread2 = do Thread::start {
                 let sched2 = sched2_cell.take();
                 sched2.run();
             };
+
+            thread1.join();
+            thread2.join();
         }
     }
 
diff --git a/src/libstd/rt/test.rs b/src/libstd/rt/test.rs
index feae8b5ffd8..260903cbcbf 100644
--- a/src/libstd/rt/test.rs
+++ b/src/libstd/rt/test.rs
@@ -125,7 +125,9 @@ pub fn run_in_mt_newsched_task(f: ~fn()) {
         }
 
         // Wait for schedulers
-        let _threads = threads;
+        for threads.consume_iter().advance() |thread| {
+            thread.join();
+        }
     }
 
 }
diff --git a/src/libstd/rt/thread.rs b/src/libstd/rt/thread.rs
index 98d08c060e0..ed0137d3b0f 100644
--- a/src/libstd/rt/thread.rs
+++ b/src/libstd/rt/thread.rs
@@ -16,7 +16,8 @@ type raw_thread = libc::c_void;
 
 pub struct Thread {
     main: ~fn(),
-    raw_thread: *raw_thread
+    raw_thread: *raw_thread,
+    joined: bool
 }
 
 impl Thread {
@@ -27,18 +28,28 @@ impl Thread {
         let raw = substart(&main);
         Thread {
             main: main,
-            raw_thread: raw
+            raw_thread: raw,
+            joined: false
         }
     }
+
+    pub fn join(self) {
+        assert!(!self.joined);
+        let mut this = self;
+        unsafe { rust_raw_thread_join(this.raw_thread); }
+        this.joined = true;
+    }
 }
 
 impl Drop for Thread {
     fn drop(&self) {
-        unsafe { rust_raw_thread_join_delete(self.raw_thread) }
+        assert!(self.joined);
+        unsafe { rust_raw_thread_delete(self.raw_thread) }
     }
 }
 
 extern {
     pub unsafe fn rust_raw_thread_start(f: &(~fn())) -> *raw_thread;
-    pub unsafe fn rust_raw_thread_join_delete(thread: *raw_thread);
+    pub unsafe fn rust_raw_thread_join(thread: *raw_thread);
+    pub unsafe fn rust_raw_thread_delete(thread: *raw_thread);
 }
diff --git a/src/libstd/rt/uv/async.rs b/src/libstd/rt/uv/async.rs
index 47e0a240e45..d0ca38317cb 100644
--- a/src/libstd/rt/uv/async.rs
+++ b/src/libstd/rt/uv/async.rs
@@ -94,12 +94,13 @@ mod test {
             let mut loop_ = Loop::new();
             let watcher = AsyncWatcher::new(&mut loop_, |w, _| w.close(||()) );
             let watcher_cell = Cell::new(watcher);
-            let _thread = do Thread::start {
+            let thread = do Thread::start {
                 let mut watcher = watcher_cell.take();
                 watcher.send();
             };
             loop_.run();
             loop_.close();
+            thread.join();
         }
     }
 }
diff --git a/src/libstd/rt/uv/net.rs b/src/libstd/rt/uv/net.rs
index a039f3ab7ed..8ea4a197269 100644
--- a/src/libstd/rt/uv/net.rs
+++ b/src/libstd/rt/uv/net.rs
@@ -715,7 +715,7 @@ mod test {
                 }
             }
 
-            let _client_thread = do Thread::start {
+            let client_thread = do Thread::start {
                 rtdebug!("starting client thread");
                 let mut loop_ = Loop::new();
                 let mut tcp_watcher = { TcpWatcher::new(&mut loop_) };
@@ -739,6 +739,7 @@ mod test {
             let mut loop_ = loop_;
             loop_.run();
             loop_.close();
+            client_thread.join();
         }
     }
 
@@ -790,7 +791,7 @@ mod test {
                 }
             }
 
-            let _client_thread = do Thread::start {
+            let client_thread = do Thread::start {
                 rtdebug!("starting client thread");
                 let mut loop_ = Loop::new();
                 let mut tcp_watcher = { TcpWatcher::new(&mut loop_) };
@@ -814,6 +815,7 @@ mod test {
             let mut loop_ = loop_;
             loop_.run();
             loop_.close();
+            client_thread.join();
         }
     }
 
@@ -855,7 +857,7 @@ mod test {
                 server.close(||{});
             }
 
-            do Thread::start {
+            let thread = do Thread::start {
                 let mut loop_ = Loop::new();
                 let mut client = UdpWatcher::new(&loop_);
                 assert!(client.bind(client_addr).is_ok());
@@ -873,6 +875,7 @@ mod test {
 
             loop_.run();
             loop_.close();
+            thread.join();
         }
     }
 
@@ -914,7 +917,7 @@ mod test {
                 server.close(||{});
             }
 
-            do Thread::start {
+            let thread = do Thread::start {
                 let mut loop_ = Loop::new();
                 let mut client = UdpWatcher::new(&loop_);
                 assert!(client.bind(client_addr).is_ok());
@@ -932,6 +935,7 @@ mod test {
 
             loop_.run();
             loop_.close();
+            thread.join();
         }
     }
 }
diff --git a/src/libstd/rt/uv/uvio.rs b/src/libstd/rt/uv/uvio.rs
index 5397b5f2c5c..53ccd20186d 100644
--- a/src/libstd/rt/uv/uvio.rs
+++ b/src/libstd/rt/uv/uvio.rs
@@ -222,11 +222,12 @@ mod test_remote {
                 };
                 remote_cell.put_back(remote);
             }
-            let _thread = do Thread::start {
+            let thread = do Thread::start {
                 remote_cell.take().fire();
             };
 
             assert!(tube.recv() == 1);
+            thread.join();
         }
     }
 }