Revise std::thread API to join by default

This commit is part of a series that introduces a `std::thread` API to
replace `std::task`.

In the new API, `spawn` returns a `JoinGuard`, which by default will
join the spawned thread when dropped. It can also be used to join
explicitly at any time, returning the thread's result. Alternatively,
the spawned thread can be explicitly detached (so no join takes place).

As part of this change, Rust processes now terminate when the main
thread exits, even if other detached threads are still running, moving
Rust closer to standard threading models. This new behavior may break code
that was relying on the previously implicit join-all.

In addition to the above, the new thread API also offers some built-in
support for building blocking abstractions in user space; see the module
doc for details.

Closes #18000

[breaking-change]

3 files changed, 44 insertions, 34 deletions

diff --git a/src/libstd/comm/blocking.rs b/src/libstd/comm/blocking.rs
index bb097265756..c477acd70aa 100644
--- a/src/libstd/comm/blocking.rs
+++ b/src/libstd/comm/blocking.rs
@@ -13,7 +13,7 @@
 use thread::Thread;
 use sync::atomic::{AtomicBool, INIT_ATOMIC_BOOL, Ordering};
 use sync::Arc;
-use kinds::marker::NoSend;
+use kinds::marker::{NoSend, NoSync};
 use mem;
 use clone::Clone;
 
@@ -30,6 +30,7 @@ pub struct SignalToken {
 pub struct WaitToken {
     inner: Arc<Inner>,
     no_send: NoSend,
+    no_sync: NoSync,
 }
 
 pub fn tokens() -> (WaitToken, SignalToken) {
@@ -40,6 +41,7 @@ pub fn tokens() -> (WaitToken, SignalToken) {
     let wait_token = WaitToken {
         inner: inner.clone(),
         no_send: NoSend,
+        no_sync: NoSync,
     };
     let signal_token = SignalToken {
         inner: inner
diff --git a/src/libstd/comm/mod.rs b/src/libstd/comm/mod.rs
index 236a055b91e..4977f966eba 100644
--- a/src/libstd/comm/mod.rs
+++ b/src/libstd/comm/mod.rs
@@ -59,26 +59,30 @@
 //! Simple usage:
 //!
 //! ```
+//! use std::thread::Thread;
+//!
 //! // Create a simple streaming channel
 //! let (tx, rx) = channel();
-//! spawn(move|| {
+//! Thread::spawn(move|| {
 //!     tx.send(10i);
-//! });
+//! }).detach();
 //! assert_eq!(rx.recv(), 10i);
 //! ```
 //!
 //! Shared usage:
 //!
 //! ```
-//! // Create a shared channel that can be sent along from many tasks
+//! use std::thread::Thread;
+//!
+//! // Create a shared channel that can be sent along from many threads
 //! // where tx is the sending half (tx for transmission), and rx is the receiving
 //! // half (rx for receiving).
 //! let (tx, rx) = channel();
 //! for i in range(0i, 10i) {
 //!     let tx = tx.clone();
-//!     spawn(move|| {
+//!     Thread::spawn(move|| {
 //!         tx.send(i);
-//!     })
+//!     }).detach()
 //! }
 //!
 //! for _ in range(0i, 10i) {
@@ -100,11 +104,13 @@
 //! Synchronous channels:
 //!
 //! ```
+//! use std::thread::Thread;
+//!
 //! let (tx, rx) = sync_channel::<int>(0);
-//! spawn(move|| {
+//! Thread::spawn(move|| {
 //!     // This will wait for the parent task to start receiving
 //!     tx.send(53);
-//! });
+//! }).detach();
 //! rx.recv();
 //! ```
 //!
@@ -451,15 +457,17 @@ impl<T> UnsafeFlavor<T> for Receiver<T> {
 /// # Example
 ///
 /// ```
+/// use std::thread::Thread;
+///
 /// // tx is is the sending half (tx for transmission), and rx is the receiving
 /// // half (rx for receiving).
 /// let (tx, rx) = channel();
 ///
 /// // Spawn off an expensive computation
-/// spawn(move|| {
+/// Thread::spawn(move|| {
 /// #   fn expensive_computation() {}
 ///     tx.send(expensive_computation());
-/// });
+/// }).detach();
 ///
 /// // Do some useful work for awhile
 ///
@@ -490,15 +498,17 @@ pub fn channel<T: Send>() -> (Sender<T>, Receiver<T>) {
 /// # Example
 ///
 /// ```
+/// use std::thread::Thread;
+///
 /// let (tx, rx) = sync_channel(1);
 ///
 /// // this returns immediately
 /// tx.send(1i);
 ///
-/// spawn(move|| {
+/// Thread::spawn(move|| {
 ///     // this will block until the previous message has been received
 ///     tx.send(2i);
-/// });
+/// }).detach();
 ///
 /// assert_eq!(rx.recv(), 1i);
 /// assert_eq!(rx.recv(), 2i);
@@ -1242,7 +1252,7 @@ mod test {
 
     test! { fn oneshot_single_thread_recv_chan_close() {
         // Receiving on a closed chan will panic
-        let res = Thread::with_join(move|| {
+        let res = Thread::spawn(move|| {
             let (tx, rx) = channel::<int>();
             drop(tx);
             rx.recv();
@@ -1314,7 +1324,7 @@ mod test {
         spawn(move|| {
             drop(tx);
         });
-        let res = Thread::with_join(move|| {
+        let res = Thread::spawn(move|| {
             assert!(rx.recv() == box 10);
         }).join();
         assert!(res.is_err());
@@ -1336,7 +1346,7 @@ mod test {
             spawn(move|| {
                 drop(rx);
             });
-            let _ = Thread::with_join(move|| {
+            let _ = Thread::spawn(move|| {
                 tx.send(1);
             }).join();
         }
@@ -1345,8 +1355,8 @@ mod test {
     test! { fn oneshot_multi_thread_recv_close_stress() {
         for _ in range(0, stress_factor()) {
             let (tx, rx) = channel::<int>();
-            spawn(proc() {
-                let res = Thread::with_join(move|| {
+            spawn(move|| {
+                let res = Thread::spawn(move|| {
                     rx.recv();
                 }).join();
                 assert!(res.is_err());
@@ -1664,7 +1674,7 @@ mod sync_tests {
 
     test! { fn oneshot_single_thread_recv_chan_close() {
         // Receiving on a closed chan will panic
-        let res = Thread::with_join(move|| {
+        let res = Thread::spawn(move|| {
             let (tx, rx) = sync_channel::<int>(0);
             drop(tx);
             rx.recv();
@@ -1741,7 +1751,7 @@ mod sync_tests {
         spawn(move|| {
             drop(tx);
         });
-        let res = Thread::with_join(move|| {
+        let res = Thread::spawn(move|| {
             assert!(rx.recv() == box 10);
         }).join();
         assert!(res.is_err());
@@ -1763,7 +1773,7 @@ mod sync_tests {
             spawn(move|| {
                 drop(rx);
             });
-            let _ = Thread::with_join(move || {
+            let _ = Thread::spawn(move || {
                 tx.send(1);
             }).join();
         }
@@ -1772,8 +1782,8 @@ mod sync_tests {
     test! { fn oneshot_multi_thread_recv_close_stress() {
         for _ in range(0, stress_factor()) {
             let (tx, rx) = sync_channel::<int>(0);
-            spawn(proc() {
-                let res = Thread::with_join(move|| {
+            spawn(move|| {
+                let res = Thread::spawn(move|| {
                     rx.recv();
                 }).join();
                 assert!(res.is_err());
diff --git a/src/libstd/comm/sync.rs b/src/libstd/comm/sync.rs
index b24c6d21fba..f75186e70e3 100644
--- a/src/libstd/comm/sync.rs
+++ b/src/libstd/comm/sync.rs
@@ -108,20 +108,18 @@ fn wait<'a, 'b, T: Send>(lock: &'a Mutex<State<T>>,
                          f: fn(SignalToken) -> Blocker)
                          -> MutexGuard<'a, State<T>>
 {
-    let me: Box<Task> = Local::take();
-    me.deschedule(1, |task| {
-        match mem::replace(&mut guard.blocker, f(task)) {
-            NoneBlocked => {}
-            _ => unreachable!(),
-        }
-        mem::drop(guard);
-        Ok(())
-    });
-    lock.lock()
+    let (wait_token, signal_token) = blocking::tokens();
+    match mem::replace(&mut guard.blocker, f(signal_token)) {
+        NoneBlocked => {}
+        _ => unreachable!(),
+    }
+    drop(guard);        // unlock
+    wait_token.wait();  // block
+    lock.lock()         // relock
 }
 
-/// Wakes up a task, dropping the lock at the correct time
-fn wakeup<T>(task: BlockedTask, guard: MutexGuard<State<T>>) {
+/// Wakes up a thread, dropping the lock at the correct time
+fn wakeup<T>(token: SignalToken, guard: MutexGuard<State<T>>) {
     // We need to be careful to wake up the waiting task *outside* of the mutex
     // in case it incurs a context switch.
     drop(guard);