diff options
| author | Aaron Turon <aturon@mozilla.com> | 2014-12-14 00:05:32 -0800 |
|---|---|---|
| committer | Aaron Turon <aturon@mozilla.com> | 2014-12-18 23:31:52 -0800 |
| commit | a27fbac86849e07a0a6c746869d8f78319bd3a16 (patch) | |
| tree | f17d75fcdd4d353f5ff919e491a5fc71252c0ef1 /src/libstd | |
| parent | 13f302d0c5dd3a88426da53ba07cdbe16459635b (diff) | |
| download | rust-a27fbac86849e07a0a6c746869d8f78319bd3a16.tar.gz rust-a27fbac86849e07a0a6c746869d8f78319bd3a16.zip | |
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]
Diffstat (limited to 'src/libstd')
44 files changed, 320 insertions, 1008 deletions
diff --git a/src/libstd/c_str.rs b/src/libstd/c_str.rs index 6bfb219d9c4..8fe3642e702 100644 --- a/src/libstd/c_str.rs +++ b/src/libstd/c_str.rs @@ -637,7 +637,7 @@ mod tests { #[test] fn test_to_c_str_fail() { - assert!(Thread::with_join(move|| { "he\x00llo".to_c_str() }).join().is_err()); + assert!(Thread::spawn(move|| { "he\x00llo".to_c_str() }).join().is_err()); } #[test] 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); diff --git a/src/libstd/io/comm_adapters.rs b/src/libstd/io/comm_adapters.rs index d2a9861737d..e865bf42bd0 100644 --- a/src/libstd/io/comm_adapters.rs +++ b/src/libstd/io/comm_adapters.rs @@ -167,7 +167,7 @@ mod test { tx.send(vec![3u8, 4u8]); tx.send(vec![5u8, 6u8]); tx.send(vec![7u8, 8u8]); - }); + }).detach(); let mut reader = ChanReader::new(rx); let mut buf = [0u8, ..3]; @@ -210,7 +210,7 @@ mod test { tx.send(b"rld\nhow ".to_vec()); tx.send(b"are you?".to_vec()); tx.send(b"".to_vec()); - }); + }).detach(); let mut reader = ChanReader::new(rx); @@ -229,11 +229,7 @@ mod test { writer.write_be_u32(42).unwrap(); let wanted = vec![0u8, 0u8, 0u8, 42u8]; -<<<<<<< HEAD - let got = match task::try(move|| { rx.recv() }) { -======= - let got = match Thread::with_join(proc() { rx.recv() }).join() { ->>>>>>> Fallout from new thread API + let got = match Thread::spawn(move|| { rx.recv() }).join() { Ok(got) => got, Err(_) => panic!(), }; diff --git a/src/libstd/io/mod.rs b/src/libstd/io/mod.rs index 6a6d467e86c..5807a3bc466 100644 --- a/src/libstd/io/mod.rs +++ b/src/libstd/io/mod.rs @@ -105,6 +105,7 @@ //! # #![allow(dead_code)] //! use std::io::{TcpListener, TcpStream}; //! use std::io::{Acceptor, Listener}; +//! use std::thread::Thread; //! //! let listener = TcpListener::bind("127.0.0.1:80"); //! @@ -119,10 +120,10 @@ //! for stream in acceptor.incoming() { //! match stream { //! Err(e) => { /* connection failed */ } -//! Ok(stream) => spawn(move|| { +//! Ok(stream) => Thread::spawn(move|| { //! // connection succeeded //! handle_client(stream) -//! }) +//! }).detach() //! } //! } //! diff --git a/src/libstd/io/net/tcp.rs b/src/libstd/io/net/tcp.rs index 0e9a93e4275..6bc952f30af 100644 --- a/src/libstd/io/net/tcp.rs +++ b/src/libstd/io/net/tcp.rs @@ -136,16 +136,17 @@ impl TcpStream { /// use std::io::timer; /// use std::io::TcpStream; /// use std::time::Duration; + /// use std::thread::Thread; /// /// let mut stream = TcpStream::connect("127.0.0.1:34254").unwrap(); /// let stream2 = stream.clone(); /// - /// spawn(move|| { + /// Thread::spawn(move|| { /// // close this stream after one second /// timer::sleep(Duration::seconds(1)); /// let mut stream = stream2; /// stream.close_read(); - /// }); + /// }).detach(); /// /// // wait for some data, will get canceled after one second /// let mut buf = [0]; @@ -279,6 +280,7 @@ impl sys_common::AsInner<TcpStreamImp> for TcpStream { /// # #![allow(dead_code)] /// use std::io::{TcpListener, TcpStream}; /// use std::io::{Acceptor, Listener}; +/// use std::thread::Thread; /// /// let listener = TcpListener::bind("127.0.0.1:80"); /// @@ -293,10 +295,10 @@ impl sys_common::AsInner<TcpStreamImp> for TcpStream { /// for stream in acceptor.incoming() { /// match stream { /// Err(e) => { /* connection failed */ } -/// Ok(stream) => spawn(move|| { +/// Ok(stream) => Thread::spawn(move|| { /// // connection succeeded /// handle_client(stream) -/// }) +/// }).detach() /// } /// } /// @@ -416,11 +418,12 @@ impl TcpAcceptor { /// ``` /// # #![allow(experimental)] /// use std::io::{TcpListener, Listener, Acceptor, EndOfFile}; + /// use std::thread::Thread; /// /// let mut a = TcpListener::bind("127.0.0.1:8482").listen().unwrap(); /// let a2 = a.clone(); /// - /// spawn(move|| { + /// Thread::spawn(move|| { /// let mut a2 = a2; /// for socket in a2.incoming() { /// match socket { @@ -429,7 +432,7 @@ impl TcpAcceptor { /// Err(e) => panic!("unexpected error: {}", e), /// } /// } - /// }); + /// }).detach(); /// /// # fn wait_for_sigint() {} /// // Now that our accept loop is running, wait for the program to be diff --git a/src/libstd/io/process.rs b/src/libstd/io/process.rs index d9acb94714b..60360a2bc64 100644 --- a/src/libstd/io/process.rs +++ b/src/libstd/io/process.rs @@ -698,7 +698,7 @@ impl Process { Thread::spawn(move |:| { let mut stream = stream; tx.send(stream.read_to_end()) - }); + }).detach(); } None => tx.send(Ok(Vec::new())) } diff --git a/src/libstd/lib.rs b/src/libstd/lib.rs index 5796f894661..78c194745a8 100644 --- a/src/libstd/lib.rs +++ b/src/libstd/lib.rs @@ -230,7 +230,6 @@ pub mod hash; /* Threads and communication */ pub mod task; -#[allow(missing_docs)] pub mod thread; pub mod sync; pub mod comm; diff --git a/src/libstd/macros.rs b/src/libstd/macros.rs index 798dac1a72f..3d03b5324b9 100644 --- a/src/libstd/macros.rs +++ b/src/libstd/macros.rs @@ -353,13 +353,15 @@ macro_rules! vec { /// # Example /// /// ``` +/// use std::thread::Thread; +/// /// let (tx1, rx1) = channel(); /// let (tx2, rx2) = channel(); /// # fn long_running_task() {} /// # fn calculate_the_answer() -> int { 42i } /// -/// spawn(move|| { long_running_task(); tx1.send(()) }); -/// spawn(move|| { tx2.send(calculate_the_answer()) }); +/// Thread::spawn(move|| { long_running_task(); tx1.send(()) }).detach(); +/// Thread::spawn(move|| { tx2.send(calculate_the_answer()) }).detach(); /// /// select! ( /// () = rx1.recv() => println!("the long running task finished first"), diff --git a/src/libstd/os.rs b/src/libstd/os.rs index 9ec3128302c..a049ea01b6d 100644 --- a/src/libstd/os.rs +++ b/src/libstd/os.rs @@ -26,6 +26,7 @@ #![allow(missing_docs)] #![allow(non_snake_case)] +#![allow(unused_imports)] pub use self::MemoryMapKind::*; pub use self::MapOption::*; @@ -37,7 +38,7 @@ use fmt; use io::{IoResult, IoError}; use iter::{Iterator, IteratorExt}; use kinds::Copy; -use libc::{c_void, c_int}; +use libc::{c_void, c_int, c_char}; use libc; use boxed::Box; use ops::{Drop, FnOnce}; diff --git a/src/libstd/path/posix.rs b/src/libstd/path/posix.rs index 5ab8eb6b942..62f64159c04 100644 --- a/src/libstd/path/posix.rs +++ b/src/libstd/path/posix.rs @@ -515,17 +515,17 @@ mod tests { #[test] fn test_null_byte() { use thread::Thread; - let result = Thread::with_join(move|| { + let result = Thread::spawn(move|| { Path::new(b"foo/bar\0") }).join(); assert!(result.is_err()); - let result = Thread::with_join(move|| { + let result = Thread::spawn(move|| { Path::new("test").set_filename(b"f\0o") }).join(); assert!(result.is_err()); - let result = Thread::with_join(move|| { + let result = Thread::spawn(move|| { Path::new("test").push(b"f\0o"); }).join(); assert!(result.is_err()); diff --git a/src/libstd/path/windows.rs b/src/libstd/path/windows.rs index 171707c10c5..5cbefb0d3d8 100644 --- a/src/libstd/path/windows.rs +++ b/src/libstd/path/windows.rs @@ -1299,17 +1299,17 @@ mod tests { #[test] fn test_null_byte() { use thread::Thread; - let result = Thread::with_join(move|| { + let result = Thread::spawn(move|| { Path::new(b"foo/bar\0") }).join(); assert!(result.is_err()); - let result = Thread::with_join(move|| { + let result = Thread::spawn(move|| { Path::new("test").set_filename(b"f\0o") }).join(); assert!(result.is_err()); - let result = Thread::with_join(move|| { + let result = Thread::spawn(move || { Path::new("test").push(b"f\0o"); }).join(); assert!(result.is_err()); diff --git a/src/libstd/rand/os.rs b/src/libstd/rand/os.rs index d2ae66d6578..dbe5f919329 100644 --- a/src/libstd/rand/os.rs +++ b/src/libstd/rand/os.rs @@ -379,7 +379,7 @@ mod test { r.fill_bytes(&mut v); Thread::yield_now(); } - }); + }).detach(); } // start all the tasks diff --git a/src/libstd/rt/at_exit_imp.rs b/src/libstd/rt/at_exit_imp.rs index 1b97a01146c..b63b4ced005 100644 --- a/src/libstd/rt/at_exit_imp.rs +++ b/src/libstd/rt/at_exit_imp.rs @@ -45,7 +45,7 @@ pub fn cleanup() { let queue: Box<Queue> = mem::transmute(queue); let v = mem::replace(&mut *queue.lock(), Vec::new()); for to_run in v.into_iter() { - to_run.invoke(); + to_run.invoke(()); } } } diff --git a/src/libstd/rt/local.rs b/src/libstd/rt/local.rs deleted file mode 100644 index 089960a6bc8..00000000000 --- a/src/libstd/rt/local.rs +++ /dev/null @@ -1,131 +0,0 @@ -// 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 core::prelude::*; - -use boxed::Box; -use rt::local_ptr; -use rt::task::Task; - -/// Encapsulates some task-local data. -pub trait Local<Borrowed> { - fn put(value: Box<Self>); - fn take() -> Box<Self>; - fn try_take() -> Option<Box<Self>>; - fn exists(unused_value: Option<Self>) -> bool; - fn borrow(unused_value: Option<Self>) -> Borrowed; - unsafe fn unsafe_take() -> Box<Self>; - unsafe fn unsafe_borrow() -> *mut Self; - unsafe fn try_unsafe_borrow() -> Option<*mut Self>; -} - -impl Local<local_ptr::Borrowed<Task>> for Task { - #[inline] - fn put(value: Box<Task>) { unsafe { local_ptr::put(value) } } - #[inline] - fn take() -> Box<Task> { unsafe { local_ptr::take() } } - #[inline] - fn try_take() -> Option<Box<Task>> { unsafe { local_ptr::try_take() } } - fn exists(_: Option<Task>) -> bool { local_ptr::exists() } - #[inline] - fn borrow(_: Option<Task>) -> local_ptr::Borrowed<Task> { - unsafe { - local_ptr::borrow::<Task>() - } - } - #[inline] - unsafe fn unsafe_take() -> Box<Task> { local_ptr::unsafe_take() } - #[inline] - unsafe fn unsafe_borrow() -> *mut Task { local_ptr::unsafe_borrow() } - #[inline] - unsafe fn try_unsafe_borrow() -> Option<*mut Task> { - local_ptr::try_unsafe_borrow() - } -} - -#[cfg(test)] -mod test { - use prelude::*; - use super::*; - use super::super::thread::Thread; - use super::super::task::Task; - - #[test] - fn thread_local_task_smoke_test() { - Thread::start(move|| { - let task = box Task::new(None, None); - Local::put(task); - let task: Box<Task> = Local::take(); - cleanup_task(task); - }).join(); - } - - #[test] - fn thread_local_task_two_instances() { - Thread::start(move|| { - let task = box Task::new(None, None); - Local::put(task); - let task: Box<Task> = Local::take(); - cleanup_task(task); - let task = box Task::new(None, None); - Local::put(task); - let task: Box<Task> = Local::take(); - cleanup_task(task); - }).join(); - } - - #[test] - fn borrow_smoke_test() { - Thread::start(move|| { - let task = box Task::new(None, None); - Local::put(task); - - unsafe { - let _task: *mut Task = Local::unsafe_borrow(); - } - let task: Box<Task> = Local::take(); - cleanup_task(task); - }).join(); - } - - #[test] - fn borrow_with_return() { - Thread::start(move|| { - let task = box Task::new(None, None); - Local::put(task); - - { - let _ = Local::borrow(None::<Task>); - } - - let task: Box<Task> = Local::take(); - cleanup_task(task); - }).join(); - } - - #[test] - fn try_take() { - Thread::start(move|| { - let task = box Task::new(None, None); - Local::put(task); - - let t: Box<Task> = Local::try_take().unwrap(); - let u: Option<Box<Task>> = Local::try_take(); - assert!(u.is_none()); - - cleanup_task(t); - }).join(); - } - - fn cleanup_task(t: Box<Task>) { - t.drop(); - } - -} diff --git a/src/libstd/rt/mod.rs b/src/libstd/rt/mod.rs index acc05cbf140..fd50d845716 100644 --- a/src/libstd/rt/mod.rs +++ b/src/libstd/rt/mod.rs @@ -27,6 +27,7 @@ use os; use thunk::Thunk; use kinds::Send; use thread::Thread; +use ops::FnOnce; use sys; use sys_common; use sys_common::thread_info::{mod, NewThread}; @@ -145,8 +146,8 @@ fn lang_start(main: *const u8, argc: int, argv: *const *const u8) -> int { /// /// It is forbidden for procedures to register more `at_exit` handlers when they /// are running, and doing so will lead to a process abort. -pub fn at_exit(f: proc():Send) { - at_exit_imp::push(f); +pub fn at_exit<F:FnOnce()+Send>(f: F) { + at_exit_imp::push(Thunk::new(f)); } /// One-time runtime cleanup. diff --git a/src/libstd/rt/mutex.rs b/src/libstd/rt/mutex.rs deleted file mode 100644 index 381f14570df..00000000000 --- a/src/libstd/rt/mutex.rs +++ /dev/null @@ -1,406 +0,0 @@ -// Copyright 2013-2014 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. - -//! A native mutex and condition variable type. -//! -//! This module contains bindings to the platform's native mutex/condition -//! variable primitives. It provides two types: `StaticNativeMutex`, which can -//! be statically initialized via the `NATIVE_MUTEX_INIT` value, and a simple -//! wrapper `NativeMutex` that has a destructor to clean up after itself. These -//! objects serve as both mutexes and condition variables simultaneously. -//! -//! The static lock is lazily initialized, but it can only be unsafely -//! destroyed. A statically initialized lock doesn't necessarily have a time at -//! which it can get deallocated. For this reason, there is no `Drop` -//! implementation of the static mutex, but rather the `destroy()` method must -//! be invoked manually if destruction of the mutex is desired. -//! -//! The non-static `NativeMutex` type does have a destructor, but cannot be -//! statically initialized. -//! -//! It is not recommended to use this type for idiomatic rust use. These types -//! are appropriate where no other options are available, but other rust -//! concurrency primitives should be used before them: the `sync` crate defines -//! `StaticMutex` and `Mutex` types. -//! -//! # Example -//! -//! ```rust -//! use rt::mutex::{NativeMutex, StaticNativeMutex, NATIVE_MUTEX_INIT}; -//! -//! // Use a statically initialized mutex -//! static LOCK: StaticNativeMutex = NATIVE_MUTEX_INIT; -//! -//! unsafe { -//! let _guard = LOCK.lock(); -//! } // automatically unlocked here -//! -//! // Use a normally initialized mutex -//! unsafe { -//! let mut lock = NativeMutex::new(); -//! -//! { -//! let _guard = lock.lock(); -//! } // unlocked here -//! -//! // sometimes the RAII guard isn't appropriate -//! lock.lock_noguard(); -//! lock.unlock_noguard(); -//! } // `lock` is deallocated here -//! ``` - -#![allow(non_camel_case_types)] - -use core::prelude::*; - -use sys::mutex as imp; - -/// A native mutex suitable for storing in statics (that is, it has -/// the `destroy` method rather than a destructor). -/// -/// Prefer the `NativeMutex` type where possible, since that does not -/// require manual deallocation. -pub struct StaticNativeMutex { - inner: imp::Mutex, -} - -/// A native mutex with a destructor for clean-up. -/// -/// See `StaticNativeMutex` for a version that is suitable for storing in -/// statics. -pub struct NativeMutex { - inner: StaticNativeMutex -} - -/// Automatically unlocks the mutex that it was created from on -/// destruction. -/// -/// Using this makes lock-based code resilient to unwinding/task -/// panic, because the lock will be automatically unlocked even -/// then. -#[must_use] -pub struct LockGuard<'a> { - lock: &'a StaticNativeMutex -} - -pub const NATIVE_MUTEX_INIT: StaticNativeMutex = StaticNativeMutex { - inner: imp::MUTEX_INIT, -}; - -impl StaticNativeMutex { - /// Creates a new mutex. - /// - /// Note that a mutex created in this way needs to be explicit - /// freed with a call to `destroy` or it will leak. - /// Also it is important to avoid locking until mutex has stopped moving - pub unsafe fn new() -> StaticNativeMutex { - StaticNativeMutex { inner: imp::Mutex::new() } - } - - /// Acquires this lock. This assumes that the current thread does not - /// already hold the lock. - /// - /// # Example - /// - /// ```rust - /// use rt::mutex::{StaticNativeMutex, NATIVE_MUTEX_INIT}; - /// static LOCK: StaticNativeMutex = NATIVE_MUTEX_INIT; - /// unsafe { - /// let _guard = LOCK.lock(); - /// // critical section... - /// } // automatically unlocked in `_guard`'s destructor - /// ``` - /// - /// # Unsafety - /// - /// This method is unsafe because it will not function correctly if this - /// mutex has been *moved* since it was last used. The mutex can move an - /// arbitrary number of times before its first usage, but once a mutex has - /// been used once it is no longer allowed to move (or otherwise it invokes - /// undefined behavior). - /// - /// Additionally, this type does not take into account any form of - /// scheduling model. This will unconditionally block the *os thread* which - /// is not always desired. - pub unsafe fn lock<'a>(&'a self) -> LockGuard<'a> { - self.inner.lock(); - - LockGuard { lock: self } - } - - /// Attempts to acquire the lock. The value returned is `Some` if - /// the attempt succeeded. - /// - /// # Unsafety - /// - /// This method is unsafe for the same reasons as `lock`. - pub unsafe fn trylock<'a>(&'a self) -> Option<LockGuard<'a>> { - if self.inner.trylock() { - Some(LockGuard { lock: self }) - } else { - None - } - } - - /// Acquire the lock without creating a `LockGuard`. - /// - /// These needs to be paired with a call to `.unlock_noguard`. Prefer using - /// `.lock`. - /// - /// # Unsafety - /// - /// This method is unsafe for the same reasons as `lock`. Additionally, this - /// does not guarantee that the mutex will ever be unlocked, and it is - /// undefined to drop an already-locked mutex. - pub unsafe fn lock_noguard(&self) { self.inner.lock() } - - /// Attempts to acquire the lock without creating a - /// `LockGuard`. The value returned is whether the lock was - /// acquired or not. - /// - /// If `true` is returned, this needs to be paired with a call to - /// `.unlock_noguard`. Prefer using `.trylock`. - /// - /// # Unsafety - /// - /// This method is unsafe for the same reasons as `lock_noguard`. - pub unsafe fn trylock_noguard(&self) -> bool { - self.inner.trylock() - } - - /// Unlocks the lock. This assumes that the current thread already holds the - /// lock. - /// - /// # Unsafety - /// - /// This method is unsafe for the same reasons as `lock`. Additionally, it - /// is not guaranteed that this is unlocking a previously locked mutex. It - /// is undefined to unlock an unlocked mutex. - pub unsafe fn unlock_noguard(&self) { self.inner.unlock() } - - /// Block on the internal condition variable. - /// - /// This function assumes that the lock is already held. Prefer - /// using `LockGuard.wait` since that guarantees that the lock is - /// held. - /// - /// # Unsafety - /// - /// This method is unsafe for the same reasons as `lock`. Additionally, this - /// is unsafe because the mutex may not be currently locked. - pub unsafe fn wait_noguard(&self) { self.inner.wait() } - - /// Signals a thread in `wait` to wake up - /// - /// # Unsafety - /// - /// This method is unsafe for the same reasons as `lock`. Additionally, this - /// is unsafe because the mutex may not be currently locked. - pub unsafe fn signal_noguard(&self) { self.inner.signal() } - - /// This function is especially unsafe because there are no guarantees made - /// that no other thread is currently holding the lock or waiting on the - /// condition variable contained inside. - pub unsafe fn destroy(&self) { self.inner.destroy() } -} - -impl NativeMutex { - /// Creates a new mutex. - /// - /// The user must be careful to ensure the mutex is not locked when its is - /// being destroyed. - /// Also it is important to avoid locking until mutex has stopped moving - pub unsafe fn new() -> NativeMutex { - NativeMutex { inner: StaticNativeMutex::new() } - } - - /// Acquires this lock. This assumes that the current thread does not - /// already hold the lock. - /// - /// # Example - /// - /// ```rust - /// use rt::mutex::NativeMutex; - /// unsafe { - /// let mut lock = NativeMutex::new(); - /// - /// { - /// let _guard = lock.lock(); - /// // critical section... - /// } // automatically unlocked in `_guard`'s destructor - /// } - /// ``` - /// - /// # Unsafety - /// - /// This method is unsafe due to the same reasons as - /// `StaticNativeMutex::lock`. - pub unsafe fn lock<'a>(&'a self) -> LockGuard<'a> { - self.inner.lock() - } - - /// Attempts to acquire the lock. The value returned is `Some` if - /// the attempt succeeded. - /// - /// # Unsafety - /// - /// This method is unsafe due to the same reasons as - /// `StaticNativeMutex::trylock`. - pub unsafe fn trylock<'a>(&'a self) -> Option<LockGuard<'a>> { - self.inner.trylock() - } - - /// Acquire the lock without creating a `LockGuard`. - /// - /// These needs to be paired with a call to `.unlock_noguard`. Prefer using - /// `.lock`. - /// - /// # Unsafety - /// - /// This method is unsafe due to the same reasons as - /// `StaticNativeMutex::lock_noguard`. - pub unsafe fn lock_noguard(&self) { self.inner.lock_noguard() } - - /// Attempts to acquire the lock without creating a - /// `LockGuard`. The value returned is whether the lock was - /// acquired or not. - /// - /// If `true` is returned, this needs to be paired with a call to - /// `.unlock_noguard`. Prefer using `.trylock`. - /// - /// # Unsafety - /// - /// This method is unsafe due to the same reasons as - /// `StaticNativeMutex::trylock_noguard`. - pub unsafe fn trylock_noguard(&self) -> bool { - self.inner.trylock_noguard() - } - - /// Unlocks the lock. This assumes that the current thread already holds the - /// lock. - /// - /// # Unsafety - /// - /// This method is unsafe due to the same reasons as - /// `StaticNativeMutex::unlock_noguard`. - pub unsafe fn unlock_noguard(&self) { self.inner.unlock_noguard() } - - /// Block on the internal condition variable. - /// - /// This function assumes that the lock is already held. Prefer - /// using `LockGuard.wait` since that guarantees that the lock is - /// held. - /// - /// # Unsafety - /// - /// This method is unsafe due to the same reasons as - /// `StaticNativeMutex::wait_noguard`. - pub unsafe fn wait_noguard(&self) { self.inner.wait_noguard() } - - /// Signals a thread in `wait` to wake up - /// - /// # Unsafety - /// - /// This method is unsafe due to the same reasons as - /// `StaticNativeMutex::signal_noguard`. - pub unsafe fn signal_noguard(&self) { self.inner.signal_noguard() } -} - -impl Drop for NativeMutex { - fn drop(&mut self) { - unsafe {self.inner.destroy()} - } -} - -impl<'a> LockGuard<'a> { - /// Block on the internal condition variable. - pub unsafe fn wait(&self) { - self.lock.wait_noguard() - } - - /// Signals a thread in `wait` to wake up. - pub unsafe fn signal(&self) { - self.lock.signal_noguard() - } -} - -#[unsafe_destructor] -impl<'a> Drop for LockGuard<'a> { - fn drop(&mut self) { - unsafe {self.lock.unlock_noguard()} - } -} - -#[cfg(test)] -mod test { - use prelude::*; - - use mem::drop; - use super::{StaticNativeMutex, NATIVE_MUTEX_INIT}; - use rt::thread::Thread; - - #[test] - fn smoke_lock() { - static LK: StaticNativeMutex = NATIVE_MUTEX_INIT; - unsafe { - let _guard = LK.lock(); - } - } - - #[test] - fn smoke_cond() { - static LK: StaticNativeMutex = NATIVE_MUTEX_INIT; - unsafe { - let guard = LK.lock(); - let t = Thread::start(move|| { - let guard = LK.lock(); - guard.signal(); - }); - guard.wait(); - drop(guard); - - t.join(); - } - } - - #[test] - fn smoke_lock_noguard() { - static LK: StaticNativeMutex = NATIVE_MUTEX_INIT; - unsafe { - LK.lock_noguard(); - LK.unlock_noguard(); - } - } - - #[test] - fn smoke_cond_noguard() { - static LK: StaticNativeMutex = NATIVE_MUTEX_INIT; - unsafe { - LK.lock_noguard(); - let t = Thread::start(move|| { - LK.lock_noguard(); - LK.signal_noguard(); - LK.unlock_noguard(); - }); - LK.wait_noguard(); - LK.unlock_noguard(); - - t.join(); - } - } - - #[test] - fn destroy_immediately() { - unsafe { - let m = StaticNativeMutex::new(); - m.destroy(); - } - } -} diff --git a/src/libstd/rt/thread.rs b/src/libstd/rt/thread.rs deleted file mode 100644 index 9eb7048f1e6..00000000000 --- a/src/libstd/rt/thread.rs +++ /dev/null @@ -1,170 +0,0 @@ -// Copyright 2013-2014 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. - -//! Native os-thread management -//! -//! This modules contains bindings necessary for managing OS-level threads. -//! These functions operate outside of the rust runtime, creating threads -//! which are not used for scheduling in any way. - -#![allow(non_camel_case_types)] - -use core::prelude::*; - -use boxed::Box; -use mem; -use sys::stack_overflow; -use sys::thread as imp; - -pub unsafe fn init() { - imp::guard::init(); - stack_overflow::init(); -} - -pub unsafe fn cleanup() { - stack_overflow::cleanup(); -} - -/// This struct represents a native thread's state. This is used to join on an -/// existing thread created in the join-able state. -pub struct Thread<T> { - native: imp::rust_thread, - joined: bool, - packet: Box<Option<T>>, -} - -static DEFAULT_STACK_SIZE: uint = 1024 * 1024; - -/// Returns the last writable byte of the main thread's stack next to the guard -/// page. Must be called from the main thread. -pub fn main_guard_page() -> uint { - unsafe { - imp::guard::main() - } -} - -/// Returns the last writable byte of the current thread's stack next to the -/// guard page. Must not be called from the main thread. -pub fn current_guard_page() -> uint { - unsafe { - imp::guard::current() - } -} - -// There are two impl blocks b/c if T were specified at the top then it's just a -// pain to specify a type parameter on Thread::spawn (which doesn't need the -// type parameter). -impl Thread<()> { - /// Starts execution of a new OS thread. - /// - /// This function will not wait for the thread to join, but a handle to the - /// thread will be returned. - /// - /// Note that the handle returned is used to acquire the return value of the - /// procedure `main`. The `join` function will wait for the thread to finish - /// and return the value that `main` generated. - /// - /// Also note that the `Thread` returned will *always* wait for the thread - /// to finish executing. This means that even if `join` is not explicitly - /// called, when the `Thread` falls out of scope its destructor will block - /// waiting for the OS thread. - pub fn start<T: Send>(main: proc():Send -> T) -> Thread<T> { - Thread::start_stack(DEFAULT_STACK_SIZE, main) - } - - /// Performs the same functionality as `start`, but specifies an explicit - /// stack size for the new thread. - pub fn start_stack<T: Send>(stack: uint, main: proc():Send -> T) -> Thread<T> { - - // We need the address of the packet to fill in to be stable so when - // `main` fills it in it's still valid, so allocate an extra box to do - // so. - let packet = box None; - let packet2: *mut Option<T> = unsafe { - *mem::transmute::<&Box<Option<T>>, *const *mut Option<T>>(&packet) - }; - let main = proc() unsafe { *packet2 = Some(main()); }; - let native = unsafe { imp::create(stack, box main) }; - - Thread { - native: native, - joined: false, - packet: packet, - } - } - - /// This will spawn a new thread, but it will not wait for the thread to - /// finish, nor is it possible to wait for the thread to finish. - /// - /// This corresponds to creating threads in the 'detached' state on unix - /// systems. Note that platforms may not keep the main program alive even if - /// there are detached thread still running around. - pub fn spawn(main: proc():Send) { - Thread::spawn_stack(DEFAULT_STACK_SIZE, main) - } - - /// Performs the same functionality as `spawn`, but explicitly specifies a - /// stack size for the new thread. - pub fn spawn_stack(stack: uint, main: proc():Send) { - unsafe { - let handle = imp::create(stack, box main); - imp::detach(handle); - } - } - - /// Relinquishes the CPU slot that this OS-thread is currently using, - /// allowing another thread to run for awhile. - pub fn yield_now() { - unsafe { imp::yield_now(); } - } -} - -impl<T: Send> Thread<T> { - /// Wait for this thread to finish, returning the result of the thread's - /// calculation. - pub fn join(mut self) -> T { - assert!(!self.joined); - unsafe { imp::join(self.native) }; - self.joined = true; - assert!(self.packet.is_some()); - self.packet.take().unwrap() - } -} - -#[unsafe_destructor] -impl<T: Send> Drop for Thread<T> { - fn drop(&mut self) { - // This is required for correctness. If this is not done then the thread - // would fill in a return box which no longer exists. - if !self.joined { - unsafe { imp::join(self.native) }; - } - } -} - -#[cfg(test)] -mod tests { - use super::Thread; - - #[test] - fn smoke() { Thread::start(proc (){}).join(); } - - #[test] - fn data() { assert_eq!(Thread::start(proc () { 1i }).join(), 1); } - - #[test] - fn detached() { Thread::spawn(proc () {}) } - - #[test] - fn small_stacks() { - assert_eq!(42i, Thread::start_stack(0, proc () 42i).join()); - assert_eq!(42i, Thread::start_stack(1, proc () 42i).join()); - } -} diff --git a/src/libstd/rt/unwind.rs b/src/libstd/rt/unwind.rs index c896f4e39da..8ef10cbbd77 100644 --- a/src/libstd/rt/unwind.rs +++ b/src/libstd/rt/unwind.rs @@ -373,6 +373,7 @@ pub mod eabi { pub use self::EXCEPTION_DISPOSITION::*; use rt::libunwind as uw; use libc::{c_void, c_int}; + use kinds::Copy; #[repr(C)] #[allow(missing_copy_implementations)] diff --git a/src/libstd/rt/util.rs b/src/libstd/rt/util.rs index 86dbb6066f3..fa527a70f83 100644 --- a/src/libstd/rt/util.rs +++ b/src/libstd/rt/util.rs @@ -89,6 +89,7 @@ pub fn default_sched_threads() -> uint { pub const ENFORCE_SANITY: bool = true || !cfg!(rtopt) || cfg!(rtdebug) || cfg!(rtassert); +#[allow(missing_copy_implementations)] pub struct Stdio(libc::c_int); #[allow(non_upper_case_globals)] diff --git a/src/libstd/sync/atomic.rs b/src/libstd/sync/atomic.rs index fe5b962fa4b..26778ef70b3 100644 --- a/src/libstd/sync/atomic.rs +++ b/src/libstd/sync/atomic.rs @@ -42,14 +42,15 @@ //! ``` //! use std::sync::Arc; //! use std::sync::atomic::{AtomicUint, SeqCst}; +//! use std::thread::Thread; //! //! fn main() { //! let spinlock = Arc::new(AtomicUint::new(1)); //! //! let spinlock_clone = spinlock.clone(); -//! spawn(move|| { +//! Thread::spawn(move|| { //! spinlock_clone.store(0, SeqCst); -//! }); +//! }).detach(); //! //! // Wait for the other task to release the lock //! while spinlock.load(SeqCst) != 0 {} @@ -61,6 +62,7 @@ //! ``` //! use std::sync::Arc; //! use std::sync::atomic::{AtomicOption, SeqCst}; +//! use std::thread::Thread; //! //! fn main() { //! struct BigObject; @@ -68,14 +70,14 @@ //! let shared_big_object = Arc::new(AtomicOption::empty()); //! //! let shared_big_object_clone = shared_big_object.clone(); -//! spawn(move|| { +//! Thread::spawn(move|| { //! let unwrapped_big_object = shared_big_object_clone.take(SeqCst); //! if unwrapped_big_object.is_some() { //! println!("got a big object from another task"); //! } else { //! println!("other task hasn't sent big object yet"); //! } -//! }); +//! }).detach(); //! //! shared_big_object.swap(box BigObject, SeqCst); //! } diff --git a/src/libstd/sync/barrier.rs b/src/libstd/sync/barrier.rs index 1b8574604a0..6573d9273ce 100644 --- a/src/libstd/sync/barrier.rs +++ b/src/libstd/sync/barrier.rs @@ -15,17 +15,18 @@ use sync::{Mutex, Condvar}; /// /// ```rust /// use std::sync::{Arc, Barrier}; +/// use std::thread::Thread; /// /// let barrier = Arc::new(Barrier::new(10)); /// for _ in range(0u, 10) { /// let c = barrier.clone(); /// // The same messages will be printed together. /// // You will NOT see any interleaving. -/// spawn(move|| { +/// Thread::spawn(move|| { /// println!("before wait"); /// c.wait(); /// println!("after wait"); -/// }); +/// }).detach(); /// } /// ``` pub struct Barrier { diff --git a/src/libstd/sync/condvar.rs b/src/libstd/sync/condvar.rs index 3bdab261e19..be27c06b83c 100644 --- a/src/libstd/sync/condvar.rs +++ b/src/libstd/sync/condvar.rs @@ -36,17 +36,18 @@ use time::Duration; /// /// ``` /// use std::sync::{Arc, Mutex, Condvar}; +/// use std::thread::Thread; /// /// let pair = Arc::new((Mutex::new(false), Condvar::new())); /// let pair2 = pair.clone(); /// /// // Inside of our lock, spawn a new thread, and then wait for it to start -/// spawn(move|| { +/// Thread::spawn(move|| { /// let &(ref lock, ref cvar) = &*pair2; /// let mut started = lock.lock(); /// *started = true; /// cvar.notify_one(); -/// }); +/// }).detach(); /// /// // wait for the thread to start up /// let &(ref lock, ref cvar) = &*pair; @@ -362,4 +363,3 @@ mod tests { } } - diff --git a/src/libstd/sync/future.rs b/src/libstd/sync/future.rs index 16f2cff5998..5e9d234c642 100644 --- a/src/libstd/sync/future.rs +++ b/src/libstd/sync/future.rs @@ -142,7 +142,7 @@ impl<A:Send> Future<A> { Thread::spawn(move |:| { // Don't panic if the other end has hung up let _ = tx.send_opt(blk()); - }); + }).detach(); Future::from_receiver(rx) } diff --git a/src/libstd/sync/mutex.rs b/src/libstd/sync/mutex.rs index fc73e2957a5..4829be569cc 100644 --- a/src/libstd/sync/mutex.rs +++ b/src/libstd/sync/mutex.rs @@ -35,6 +35,7 @@ use sys_common::mutex as sys; /// /// ```rust /// use std::sync::{Arc, Mutex}; +/// use std::thread::Thread; /// const N: uint = 10; /// /// // Spawn a few threads to increment a shared variable (non-atomically), and @@ -47,7 +48,7 @@ use sys_common::mutex as sys; /// let (tx, rx) = channel(); /// for _ in range(0u, 10) { /// let (data, tx) = (data.clone(), tx.clone()); -/// spawn(move|| { +/// Thread::spawn(move|| { /// // The shared static can only be accessed once the lock is held. /// // Our non-atomic increment is safe because we're the only thread /// // which can access the shared state when the lock is held. @@ -57,7 +58,7 @@ use sys_common::mutex as sys; /// tx.send(()); /// } /// // the lock is unlocked here when `data` goes out of scope. -/// }); +/// }).detach(); /// } /// /// rx.recv(); @@ -386,7 +387,7 @@ mod test { fn test_mutex_arc_poison() { let arc = Arc::new(Mutex::new(1i)); let arc2 = arc.clone(); - let _ = Thread::with_join(move|| { + let _ = Thread::spawn(move|| { let lock = arc2.lock(); assert_eq!(*lock, 2); }).join(); @@ -414,7 +415,7 @@ mod test { fn test_mutex_arc_access_in_unwind() { let arc = Arc::new(Mutex::new(1i)); let arc2 = arc.clone(); - let _ = Thread::with_join::<()>(move|| -> () { + let _ = Thread::spawn(move|| -> () { struct Unwinder { i: Arc<Mutex<int>>, } diff --git a/src/libstd/sync/rwlock.rs b/src/libstd/sync/rwlock.rs index 1f1e9eea1d6..07b2f2cf541 100644 --- a/src/libstd/sync/rwlock.rs +++ b/src/libstd/sync/rwlock.rs @@ -409,7 +409,7 @@ mod tests { fn test_rw_arc_poison_wr() { let arc = Arc::new(RWLock::new(1i)); let arc2 = arc.clone(); - let _ = Thread::with_join(move|| { + let _ = Thread::spawn(move|| { let lock = arc2.write(); assert_eq!(*lock, 2); }).join(); @@ -422,7 +422,7 @@ mod tests { fn test_rw_arc_poison_ww() { let arc = Arc::new(RWLock::new(1i)); let arc2 = arc.clone(); - let _ = Thread::with_join(move|| { + let _ = Thread::spawn(move|| { let lock = arc2.write(); assert_eq!(*lock, 2); }).join(); @@ -434,7 +434,7 @@ mod tests { fn test_rw_arc_no_poison_rr() { let arc = Arc::new(RWLock::new(1i)); let arc2 = arc.clone(); - let _ = Thread::with_join(move|| { + let _ = Thread::spawn(move|| { let lock = arc2.read(); assert_eq!(*lock, 2); }).join(); @@ -445,7 +445,7 @@ mod tests { fn test_rw_arc_no_poison_rw() { let arc = Arc::new(RWLock::new(1i)); let arc2 = arc.clone(); - let _ = Thread::with_join(move|| { + let _ = Thread::spawn(move|| { let lock = arc2.read(); assert_eq!(*lock, 2); }).join(); @@ -468,13 +468,13 @@ mod tests { *lock = tmp + 1; } tx.send(()); - }); + }).detach(); // Readers try to catch the writer in the act let mut children = Vec::new(); for _ in range(0u, 5) { let arc3 = arc.clone(); - children.push(Thread::with_join(move|| { + children.push(Thread::spawn(move|| { let lock = arc3.read(); assert!(*lock >= 0); })); @@ -495,11 +495,7 @@ mod tests { fn test_rw_arc_access_in_unwind() { let arc = Arc::new(RWLock::new(1i)); let arc2 = arc.clone(); -<<<<<<< HEAD - let _ = task::try(move|| -> () { -======= - let _ = Thread::with_join::<()>(proc() { ->>>>>>> Fallout from new thread API + let _ = Thread::spawn(move|| -> () { struct Unwinder { i: Arc<RWLock<int>>, } diff --git a/src/libstd/sync/task_pool.rs b/src/libstd/sync/task_pool.rs index 5e7944d5fe5..793825f1b08 100644 --- a/src/libstd/sync/task_pool.rs +++ b/src/libstd/sync/task_pool.rs @@ -126,7 +126,7 @@ fn spawn_in_pool(jobs: Arc<Mutex<Receiver<Thunk>>>) { } sentinel.cancel(); - }); + }).detach(); } #[cfg(test)] diff --git a/src/libstd/sys/common/backtrace.rs b/src/libstd/sys/common/backtrace.rs index 38dc516bf3d..2ff82bfd7a9 100644 --- a/src/libstd/sys/common/backtrace.rs +++ b/src/libstd/sys/common/backtrace.rs @@ -10,8 +10,8 @@ use io::{IoResult, Writer}; use iter::{Iterator, IteratorExt}; -use option::{Some, None}; -use result::{Ok, Err}; +use option::Option::{Some, None}; +use result::Result::{Ok, Err}; use str::{StrPrelude, from_str}; use unicode::char::UnicodeChar; diff --git a/src/libstd/sys/common/helper_thread.rs b/src/libstd/sys/common/helper_thread.rs index 8aa09d9bd30..421778e2012 100644 --- a/src/libstd/sys/common/helper_thread.rs +++ b/src/libstd/sys/common/helper_thread.rs @@ -82,16 +82,12 @@ impl<M: Send> Helper<M> { *self.signal.get() = send as uint; let t = f(); -<<<<<<< HEAD - task::spawn(move |:| { -======= - Thread::spawn(proc() { ->>>>>>> Fallout from new thread API + Thread::spawn(move |:| { helper(receive, rx, t); let _g = self.lock.lock(); *self.shutdown.get() = true; self.cond.notify_one() - }); + }).detach(); rt::at_exit(move|:| { self.shutdown() }); *self.initialized.get() = true; diff --git a/src/libstd/sys/common/thread.rs b/src/libstd/sys/common/thread.rs index 5e1adfb8714..048e33399a3 100644 --- a/src/libstd/sys/common/thread.rs +++ b/src/libstd/sys/common/thread.rs @@ -14,6 +14,7 @@ use boxed::Box; use mem; use uint; use libc; +use thunk::Thunk; use sys_common::stack; use sys::{thread, stack_overflow}; @@ -26,8 +27,8 @@ pub fn start_thread(main: *mut libc::c_void) -> thread::rust_thread_return { unsafe { stack::record_os_managed_stack_bounds(0, uint::MAX); let handler = stack_overflow::Handler::new(); - let f: Box<proc()> = mem::transmute(main); - (*f)(); + let f: Box<Thunk> = mem::transmute(main); + f.invoke(()); drop(handler); mem::transmute(0 as thread::rust_thread_return) } diff --git a/src/libstd/sys/common/thread_info.rs b/src/libstd/sys/common/thread_info.rs index 206d443d294..92a896c7583 100644 --- a/src/libstd/sys/common/thread_info.rs +++ b/src/libstd/sys/common/thread_info.rs @@ -28,6 +28,11 @@ thread_local!(static THREAD_INFO: RefCell<Option<ThreadInfo>> = RefCell::new(Non impl ThreadInfo { fn with<R>(f: |&mut ThreadInfo| -> R) -> R { + if THREAD_INFO.destroyed() { + panic!("Use of std::thread::Thread::current() is not possible after \ + the thread's local data has been destroyed"); + } + THREAD_INFO.with(|c| { if c.borrow().is_none() { *c.borrow_mut() = Some(ThreadInfo { diff --git a/src/libstd/sys/unix/backtrace.rs b/src/libstd/sys/unix/backtrace.rs index d98f935076f..df9dbad2ec7 100644 --- a/src/libstd/sys/unix/backtrace.rs +++ b/src/libstd/sys/unix/backtrace.rs @@ -87,8 +87,8 @@ use c_str::CString; use io::{IoResult, Writer}; use libc; use mem; -use option::{Some, None, Option}; -use result::{Ok, Err}; +use option::Option::{mod, Some, None}; +use result::Result::{Ok, Err}; use sync::{StaticMutex, MUTEX_INIT}; use sys_common::backtrace::*; diff --git a/src/libstd/sys/unix/thread.rs b/src/libstd/sys/unix/thread.rs index 02da3a19818..2416b64f98f 100644 --- a/src/libstd/sys/unix/thread.rs +++ b/src/libstd/sys/unix/thread.rs @@ -16,6 +16,7 @@ use mem; use ptr; use libc::consts::os::posix01::{PTHREAD_CREATE_JOINABLE, PTHREAD_STACK_MIN}; use libc; +use thunk::Thunk; use sys_common::stack::RED_ZONE; use sys_common::thread::*; @@ -153,7 +154,7 @@ pub mod guard { } } -pub unsafe fn create(stack: uint, p: Box<proc():Send>) -> rust_thread { +pub unsafe fn create(stack: uint, p: Thunk) -> rust_thread { let mut native: libc::pthread_t = mem::zeroed(); let mut attr: libc::pthread_attr_t = mem::zeroed(); assert_eq!(pthread_attr_init(&mut attr), 0); @@ -181,13 +182,13 @@ pub unsafe fn create(stack: uint, p: Box<proc():Send>) -> rust_thread { }, }; - let arg: *mut libc::c_void = mem::transmute(p); + let arg: *mut libc::c_void = mem::transmute(box p); // must box since sizeof(p)=2*uint let ret = pthread_create(&mut native, &attr, thread_start, arg); assert_eq!(pthread_attr_destroy(&mut attr), 0); if ret != 0 { // be sure to not leak the closure - let _p: Box<proc():Send> = mem::transmute(arg); + let _p: Box<Box<FnOnce()+Send>> = mem::transmute(arg); panic!("failed to spawn native thread: {}", ret); } native diff --git a/src/libstd/sys/windows/backtrace.rs b/src/libstd/sys/windows/backtrace.rs index 9903d2f1ae2..b5961d783ff 100644 --- a/src/libstd/sys/windows/backtrace.rs +++ b/src/libstd/sys/windows/backtrace.rs @@ -27,9 +27,9 @@ use io::{IoResult, Writer}; use libc; use mem; use ops::Drop; -use option::{Some, None}; +use option::Option::{Some, None}; use path::Path; -use result::{Ok, Err}; +use result::Result::{Ok, Err}; use sync::{StaticMutex, MUTEX_INIT}; use slice::SliceExt; use str::StrPrelude; diff --git a/src/libstd/sys/windows/fs.rs b/src/libstd/sys/windows/fs.rs index 05be8de0b56..0fb52c758d5 100644 --- a/src/libstd/sys/windows/fs.rs +++ b/src/libstd/sys/windows/fs.rs @@ -15,7 +15,7 @@ use libc::{mod, c_int}; use c_str::CString; use mem; -use os::windoze::fill_utf16_buf_and_decode; +use sys::os::fill_utf16_buf_and_decode; use path; use ptr; use str; diff --git a/src/libstd/sys/windows/os.rs b/src/libstd/sys/windows/os.rs index 5c690180c44..2fbb9494c71 100644 --- a/src/libstd/sys/windows/os.rs +++ b/src/libstd/sys/windows/os.rs @@ -24,6 +24,9 @@ use path::{Path, GenericPath, BytesContainer}; use ptr::{mod, RawPtr}; use sync::atomic::{AtomicInt, INIT_ATOMIC_INT, SeqCst}; use sys::fs::FileDesc; +use option::Option; +use option::Option::{Some, None}; +use slice; use os::TMPBUF_SZ; use libc::types::os::arch::extra::DWORD; @@ -138,7 +141,7 @@ pub fn fill_utf16_buf_and_decode(f: |*mut u16, DWORD| -> DWORD) -> Option<String // set `res` to None and continue. let s = String::from_utf16(sub) .expect("fill_utf16_buf_and_decode: closure created invalid UTF-16"); - res = option::Some(s) + res = Some(s) } } return res; diff --git a/src/libstd/sys/windows/stack_overflow.rs b/src/libstd/sys/windows/stack_overflow.rs index 63b5b6f5863..bdf2e0bccb1 100644 --- a/src/libstd/sys/windows/stack_overflow.rs +++ b/src/libstd/sys/windows/stack_overflow.rs @@ -34,7 +34,7 @@ impl Drop for Handler { // It returns the guard page of the current task or 0 if that // guard page doesn't exist. None is returned if there's currently // no local task. -unsafe fn get_task_guard_page() -> Option<uint> { +unsafe fn get_task_guard_page() -> uint { thread_info::stack_guard() } @@ -55,9 +55,7 @@ extern "system" fn vectored_handler(ExceptionInfo: *mut EXCEPTION_POINTERS) -> L // however stack checks by limit should be disabled on Windows stack::record_sp_limit(0); - if get_task_guard_page().is_some() { - report_overflow(); - } + report_overflow(); EXCEPTION_CONTINUE_SEARCH } diff --git a/src/libstd/sys/windows/thread.rs b/src/libstd/sys/windows/thread.rs index 00f1e9767f5..4498f56c00a 100644 --- a/src/libstd/sys/windows/thread.rs +++ b/src/libstd/sys/windows/thread.rs @@ -17,6 +17,7 @@ use ptr; use libc; use libc::types::os::arch::extra::{LPSECURITY_ATTRIBUTES, SIZE_T, BOOL, LPVOID, DWORD, LPDWORD, HANDLE}; +use thunk::Thunk; use sys_common::stack::RED_ZONE; use sys_common::thread::*; @@ -43,8 +44,8 @@ pub mod guard { } } -pub unsafe fn create(stack: uint, p: Box<proc():Send>) -> rust_thread { - let arg: *mut libc::c_void = mem::transmute(p); +pub unsafe fn create(stack: uint, p: Thunk) -> rust_thread { + let arg: *mut libc::c_void = mem::transmute(box p); // FIXME On UNIX, we guard against stack sizes that are too small but // that's because pthreads enforces that stacks are at least // PTHREAD_STACK_MIN bytes big. Windows has no such lower limit, it's @@ -60,7 +61,7 @@ pub unsafe fn create(stack: uint, p: Box<proc():Send>) -> rust_thread { if ret as uint == 0 { // be sure to not leak the closure - let _p: Box<proc():Send> = mem::transmute(arg); + let _p: Box<Thunk> = mem::transmute(arg); panic!("failed to spawn native thread: {}", ret); } return ret; diff --git a/src/libstd/sys/windows/thread_local.rs b/src/libstd/sys/windows/thread_local.rs index 6c8d9639d5c..0644519aabb 100644 --- a/src/libstd/sys/windows/thread_local.rs +++ b/src/libstd/sys/windows/thread_local.rs @@ -14,8 +14,7 @@ use libc::types::os::arch::extra::{DWORD, LPVOID, BOOL}; use mem; use rt; -use rt::exclusive::Exclusive; -use sync::{ONCE_INIT, Once}; +use sync::{ONCE_INIT, Once, Mutex}; pub type Key = DWORD; pub type Dtor = unsafe extern fn(*mut u8); @@ -55,7 +54,7 @@ pub type Dtor = unsafe extern fn(*mut u8); // /threading/thread_local_storage_win.cc#L42 static INIT_DTORS: Once = ONCE_INIT; -static mut DTORS: *mut Exclusive<Vec<(Key, Dtor)>> = 0 as *mut _; +static mut DTORS: *mut Mutex<Vec<(Key, Dtor)>> = 0 as *mut _; // ------------------------------------------------------------------------- // Native bindings @@ -126,13 +125,13 @@ extern "system" { // FIXME: This could probably be at least a little faster with a BTree. fn init_dtors() { - let dtors = box Exclusive::new(Vec::<(Key, Dtor)>::new()); + let dtors = box Mutex::new(Vec::<(Key, Dtor)>::new()); unsafe { DTORS = mem::transmute(dtors); } rt::at_exit(move|| unsafe { - mem::transmute::<_, Box<Exclusive<Vec<(Key, Dtor)>>>>(DTORS); + mem::transmute::<_, Box<Mutex<Vec<(Key, Dtor)>>>>(DTORS); DTORS = 0 as *mut _; }); } diff --git a/src/libstd/task.rs b/src/libstd/task.rs index 13395fc9c76..0f08108fee5 100644 --- a/src/libstd/task.rs +++ b/src/libstd/task.rs @@ -17,21 +17,24 @@ use boxed::Box; use thread; use kinds::Send; use result::Result; +use ops::FnOnce; -/// Deprecate: use `std::thread::Cfg` instead. -#[deprecated = "use std::thread::Cfg instead"] -pub type TaskBuilder = thread::Cfg; +/// Deprecate: use `std::thread::Builder` instead. +#[deprecated = "use std::thread::Builder instead"] +pub type TaskBuilder = thread::Builder; -/// Deprecated: use `std::thread::Thread::spawn` instead. -#[deprecated = "use std::thread::Thread::spawn instead"] -pub fn spawn(f: proc(): Send) { - thread::Thread::spawn(f); +/// Deprecated: use `std::thread::Thread::spawn` and `detach` instead. +#[deprecated = "use std::thread::Thread::spawn and detach instead"] +pub fn spawn<F>(f: F) where F: FnOnce(), F: Send { + thread::Thread::spawn(f).detach(); } -/// Deprecated: use `std::thread::Thread::with_join instead`. -#[deprecated = "use std::thread::Thread::with_join instead"] -pub fn try<T: Send>(f: proc(): Send -> T) -> Result<T, Box<Any + Send>> { - thread::Thread::with_join(f).join() +/// Deprecated: use `std::thread::Thread::spawn` and `join` instead. +#[deprecated = "use std::thread::Thread::spawn and join instead"] +pub fn try<T, F>(f: F) -> Result<T, Box<Any + Send>> where + T: Send, F: FnOnce() -> T, F: Send +{ + thread::Thread::spawn(f).join() } /// Deprecated: use `std::thread::Thread::yield_now instead`. diff --git a/src/libstd/thread.rs b/src/libstd/thread.rs index 7ccfe0ab74b..6c6c0ce85eb 100644 --- a/src/libstd/thread.rs +++ b/src/libstd/thread.rs @@ -15,13 +15,12 @@ //! An executing Rust program consists of a collection of native OS threads, //! each with their own stack and local state. //! -//! Threads generally have their memory *isolated* from each other by virtue of -//! Rust's owned types (which of course may only be owned by a single thread at -//! a time). Communication between threads can be done through -//! [channels](../../std/comm/index.html), Rust's message-passing types, along -//! with [other forms of thread synchronization](../../std/sync/index.html) and -//! shared-memory data structures. In particular, types that are guaranteed to -//! be threadsafe are easily shared between threads using the +//! Communication between threads can be done through +//! [channels](../../std/comm/index.html), Rust's message-passing +//! types, along with [other forms of thread +//! synchronization](../../std/sync/index.html) and shared-memory data +//! structures. In particular, types that are guaranteed to be +//! threadsafe are easily shared between threads using the //! atomically-reference-counted container, //! [`Arc`](../../std/sync/struct.Arc.html). //! @@ -54,52 +53,43 @@ //! //! ## Spawning a thread //! -//! There are a few different ways to spawn a new thread, depending on how it -//! should relate to the parent thread. -//! -//! ### Simple detached threads -//! -//! The simplest case just spawns a completely independent (detached) thread, -//! returning a new `Thread` handle to it: +//! A new thread can be spawned using the `Thread::spawn` function: //! //! ```rust //! use std::thread::Thread; //! -//! Thread::spawn(proc() { +//! let guard = Thread::spawn(move || { //! println!("Hello, World!"); -//! }) +//! // some computation here +//! }); +//! let result = guard.join(); //! ``` //! -//! The spawned thread may outlive its parent. -//! -//! ### Joining -//! -//! Alternatively, the `with_join` constructor spawns a new thread and returns a -//! `JoinGuard` which can be used to wait until the child thread completes, -//! returning its result (or `Err` if the child thread panicked): -//! -//! ```rust -//! use std::thread::Thread; +//! The `spawn` function doesn't return a `Thread` directly; instead, it returns +//! a *join guard* from which a `Thread` can be extracted. The join guard is an +//! RAII-style guard that will automatically join the child thread (block until +//! it terminates) when it is dropped. You can join the child thread in advance +//! by calling the `join` method on the guard, which will also return the result +//! produced by the thread. //! -//! let guard = Thread::with_join(proc() { panic!() }; -//! assert!(guard.join().is_err()); -//! ``` +//! If you instead wish to *detach* the child thread, allowing it to outlive its +//! parent, you can use the `detach` method on the guard, //! -//! The guard works in RAII style, meaning that the child thread is -//! automatically joined when the guard is dropped. A handle to the thread -//! itself is available via the `thread` method on the guard. +//! A handle to the thread itself is available via the `thread` method on the +//! join guard. //! -//! ### Configured threads +//! ## Configuring threads //! -//! Finally, a new thread can be configured independently of how it is -//! spawned. Configuration is available via the `Cfg` builder, which currently -//! allows you to set the name, stack size, and writers for `println!` and -//! `panic!` for the child thread: +//! A new thread can be configured before it is spawned via the `Builder` type, +//! which currently allows you to set the name, stack size, and writers for +//! `println!` and `panic!` for the child thread: //! //! ```rust //! use std::thread; //! -//! thread::cfg().name("child1").spawn(proc() { println!("Hello, world!") }); +//! thread::Builder::new().name("child1".to_string()).spawn(move || { +//! println!("Hello, world!") +//! }).detach(); //! ``` //! //! ## Blocking support: park and unpark @@ -139,18 +129,19 @@ use core::prelude::*; use any::Any; use borrow::IntoCow; use boxed::Box; -use mem; +use cell::UnsafeCell; use sync::{Mutex, Condvar, Arc}; use string::String; use rt::{mod, unwind}; use io::{Writer, stdio}; +use thunk::Thunk; use sys::thread as imp; use sys_common::{stack, thread_info}; /// Thread configuation. Provides detailed control over the properties /// and behavior of new threads. -pub struct Cfg { +pub struct Builder { // A name for the thread-to-be, for identification in panic messages name: Option<String>, // The size of the stack for the spawned thread @@ -161,11 +152,11 @@ pub struct Cfg { stderr: Option<Box<Writer + Send>>, } -impl Cfg { +impl Builder { /// Generate the base configuration for spawning a thread, from which /// configuration methods can be chained. - pub fn new() -> Cfg { - Cfg { + pub fn new() -> Builder { + Builder { name: None, stack_size: None, stdout: None, @@ -175,41 +166,51 @@ impl Cfg { /// Name the thread-to-be. Currently the name is used for identification /// only in panic messages. - pub fn name(mut self, name: String) -> Cfg { + pub fn name(mut self, name: String) -> Builder { self.name = Some(name); self } /// Deprecated: use `name` instead #[deprecated = "use name instead"] - pub fn named<T: IntoCow<'static, String, str>>(self, name: T) -> Cfg { + pub fn named<T: IntoCow<'static, String, str>>(self, name: T) -> Builder { self.name(name.into_cow().into_owned()) } /// Set the size of the stack for the new thread. - pub fn stack_size(mut self, size: uint) -> Cfg { + pub fn stack_size(mut self, size: uint) -> Builder { self.stack_size = Some(size); self } /// Redirect thread-local stdout. #[experimental = "Will likely go away after proc removal"] - pub fn stdout(mut self, stdout: Box<Writer + Send>) -> Cfg { + pub fn stdout(mut self, stdout: Box<Writer + Send>) -> Builder { self.stdout = Some(stdout); self } /// Redirect thread-local stderr. #[experimental = "Will likely go away after proc removal"] - pub fn stderr(mut self, stderr: Box<Writer + Send>) -> Cfg { + pub fn stderr(mut self, stderr: Box<Writer + Send>) -> Builder { self.stderr = Some(stderr); self } - fn core_spawn<T: Send>(self, f: proc():Send -> T, after: proc(Result<T>):Send) - -> (imp::rust_thread, Thread) + /// Spawn a new joinable thread, and return a JoinGuard guard for it. + /// + /// See `Thead::spawn` and the module doc for more details. + pub fn spawn<T, F>(self, f: F) -> JoinGuard<T> where + T: Send, F: FnOnce() -> T, F: Send { - let Cfg { name, stack_size, stdout, stderr } = self; + self.spawn_inner(Thunk::new(f)) + } + + fn spawn_inner<T: Send>(self, f: Thunk<(), T>) -> JoinGuard<T> { + let my_packet = Arc::new(UnsafeCell::new(None)); + let their_packet = my_packet.clone(); + + let Builder { name, stack_size, stdout, stderr } = self; let stack_size = stack_size.unwrap_or(rt::min_stack()); let my_thread = Thread::new(name); @@ -221,7 +222,7 @@ impl Cfg { // because by the time that this function is executing we've already // consumed at least a little bit of stack (we don't know the exact byte // address at which our stack started). - let main = proc() { + let main = move |:| { let something_around_the_top_of_the_stack = 1; let addr = &something_around_the_top_of_the_stack as *const int; let my_stack_top = addr as uint; @@ -235,74 +236,49 @@ impl Cfg { their_thread ); - // There are two primary reasons that general try/catch is - // unsafe. The first is that we do not support nested try/catch. The - // fact that this is happening in a newly-spawned thread - // suffices. The second is that unwinding while unwinding is not - // defined. We take care of that by having an 'unwinding' flag in - // the thread itself. For these reasons, this unsafety should be ok. + let mut output = None; + let f: Thunk<(), T> = if stdout.is_some() || stderr.is_some() { + Thunk::new(move |:| { + let _ = stdout.map(stdio::set_stdout); + let _ = stderr.map(stdio::set_stderr); + f.invoke(()) + }) + } else { + f + }; + + let try_result = { + let ptr = &mut output; + + // There are two primary reasons that general try/catch is + // unsafe. The first is that we do not support nested + // try/catch. The fact that this is happening in a newly-spawned + // thread suffices. The second is that unwinding while unwinding + // is not defined. We take care of that by having an + // 'unwinding' flag in the thread itself. For these reasons, + // this unsafety should be ok. + unsafe { + unwind::try(move || *ptr = Some(f.invoke(()))) + } + }; unsafe { - let mut output = None; - let f = if stdout.is_some() || stderr.is_some() { - proc() { - let _ = stdout.map(stdio::set_stdout); - let _ = stderr.map(stdio::set_stderr); - f() - } - } else { - f - }; - - let try_result = { - let ptr = &mut output; - unwind::try(move || *ptr = Some(f())) - }; - match (output, try_result) { - (Some(data), Ok(_)) => after(Ok(data)), - (None, Err(cause)) => after(Err(cause)), + *their_packet.get() = Some(match (output, try_result) { + (Some(data), Ok(_)) => Ok(data), + (None, Err(cause)) => Err(cause), _ => unreachable!() - } + }); } }; - (unsafe { imp::create(stack_size, box main) }, my_thread) - } - - /// Spawn a detached thread, and return a handle to it. - /// - /// The new child thread may outlive its parent. - pub fn spawn(self, f: proc():Send) -> Thread { - let (native, thread) = self.core_spawn(f, proc(_) {}); - unsafe { imp::detach(native) }; - thread - } - - /// Spawn a joinable thread, and return an RAII guard for it. - pub fn with_join<T: Send>(self, f: proc():Send -> T) -> JoinGuard<T> { - // We need the address of the packet to fill in to be stable so when - // `main` fills it in it's still valid, so allocate an extra box to do - // so. - let any: Box<Any+Send> = box 0u8; // sentinel value - let my_packet = box Err(any); - let their_packet: *mut Result<T> = unsafe { - *mem::transmute::<&Box<Result<T>>, *const *mut Result<T>>(&my_packet) - }; - - let (native, thread) = self.core_spawn(f, proc(result) { - unsafe { *their_packet = result; } - }); JoinGuard { - native: native, + native: unsafe { imp::create(stack_size, Thunk::new(main)) }, joined: false, - packet: Some(my_packet), - thread: thread, + packet: my_packet, + thread: my_thread, } } } -/// A convenience function for creating configurations. -pub fn cfg() -> Cfg { Cfg::new() } - struct Inner { name: Option<String>, lock: Mutex<bool>, // true when there is a buffered unpark @@ -316,6 +292,7 @@ pub struct Thread { } impl Thread { + // Used only internally to construct a thread object without spawning fn new(name: Option<String>) -> Thread { Thread { inner: Arc::new(Inner { @@ -326,16 +303,16 @@ impl Thread { } } - /// Spawn a detached thread, and return a handle to it. + /// Spawn a new joinable thread, returning a `JoinGuard` for it. /// - /// The new child thread may outlive its parent. - pub fn spawn(f: proc():Send) -> Thread { - Cfg::new().spawn(f) - } - - /// Spawn a joinable thread, and return an RAII guard for it. - pub fn with_join<T: Send>(f: proc():Send -> T) -> JoinGuard<T> { - Cfg::new().with_join(f) + /// The join guard can be used to explicitly join the child thead (via + /// `join`), returning `Result<T>`, or it will implicitly join the child + /// upon being dropped. To detach the child, allowing it to outlive the + /// current thread, use `detach`. See the module documentation for additional details. + pub fn spawn<T, F>(f: F) -> JoinGuard<T> where + T: Send, F: FnOnce() -> T, F: Send + { + Builder::new().spawn(f) } /// Gets a handle to the thread that invokes it. @@ -353,10 +330,15 @@ impl Thread { thread_info::panicking() } - // http://cr.openjdk.java.net/~stefank/6989984.1/raw_files/new/src/os/linux/vm/os_linux.cpp /// Block unless or until the current thread's token is made available (may wake spuriously). /// /// See the module doc for more detail. + // + // The implementation currently uses the trivial strategy of a Mutex+Condvar + // with wakeup flag, which does not actually allow spurious wakeups. In the + // future, this will be implemented in a more efficient way, perhaps along the lines of + // http://cr.openjdk.java.net/~stefank/6989984.1/raw_files/new/src/os/linux/vm/os_linux.cpp + // or futuxes, and in either case may allow spurious wakeups. pub fn park() { let thread = Thread::current(); let mut guard = thread.inner.lock.lock(); @@ -394,43 +376,52 @@ impl thread_info::NewThread for Thread { pub type Result<T> = ::result::Result<T, Box<Any + Send>>; #[must_use] -/// An RAII guard that will block until thread termination when dropped. +/// An RAII-style guard that will block until thread termination when dropped. +/// +/// The type `T` is the return type for the thread's main function. pub struct JoinGuard<T> { native: imp::rust_thread, thread: Thread, joined: bool, - packet: Option<Box<Result<T>>>, + packet: Arc<UnsafeCell<Option<Result<T>>>>, } impl<T: Send> JoinGuard<T> { /// Extract a handle to the thread this guard will join on. - pub fn thread(&self) -> Thread { - self.thread.clone() + pub fn thread(&self) -> &Thread { + &self.thread } /// Wait for the associated thread to finish, returning the result of the thread's /// calculation. + /// + /// If the child thread panics, `Err` is returned with the parameter given + /// to `panic`. pub fn join(mut self) -> Result<T> { assert!(!self.joined); unsafe { imp::join(self.native) }; self.joined = true; - let box res = self.packet.take().unwrap(); - res + unsafe { + (*self.packet.get()).take().unwrap() + } + } + + /// Detaches the child thread, allowing it to outlive its parent. + pub fn detach(mut self) { + unsafe { imp::detach(self.native) }; + self.joined = true; // avoid joining in the destructor } } #[unsafe_destructor] impl<T: Send> Drop for JoinGuard<T> { fn drop(&mut self) { - // This is required for correctness. If this is not done then the thread - // would fill in a return box which no longer exists. if !self.joined { unsafe { imp::join(self.native) }; } } } -// TODO: fix tests #[cfg(test)] mod test { use any::{Any, AnyRefExt}; @@ -440,21 +431,22 @@ mod test { use result; use std::io::{ChanReader, ChanWriter}; use string::String; - use super::{Thread, cfg}; + use thunk::Thunk; + use super::{Thread, Builder}; // !!! These tests are dangerous. If something is buggy, they will hang, !!! // !!! instead of exiting cleanly. This might wedge the buildbots. !!! #[test] fn test_unnamed_thread() { - Thread::with_join(proc() { + Thread::spawn(move|| { assert!(Thread::current().name().is_none()); }).join().map_err(|_| ()).unwrap(); } #[test] fn test_named_thread() { - cfg().name("ada lovelace".to_string()).with_join(proc() { + Builder::new().name("ada lovelace".to_string()).spawn(move|| { assert!(Thread::current().name().unwrap() == "ada lovelace".to_string()); }).join().map_err(|_| ()).unwrap(); } @@ -462,15 +454,15 @@ mod test { #[test] fn test_run_basic() { let (tx, rx) = channel(); - Thread::spawn(proc() { + Thread::spawn(move|| { tx.send(()); - }); + }).detach(); rx.recv(); } #[test] fn test_join_success() { - match Thread::with_join::<String>(proc() { + match Thread::spawn(move|| -> String { "Success!".to_string() }).join().as_ref().map(|s| s.as_slice()) { result::Result::Ok("Success!") => (), @@ -480,7 +472,7 @@ mod test { #[test] fn test_join_panic() { - match Thread::with_join(proc() { + match Thread::spawn(move|| { panic!() }).join() { result::Result::Err(_) => (), @@ -496,13 +488,13 @@ mod test { fn f(i: int, tx: Sender<()>) { let tx = tx.clone(); - Thread::spawn(proc() { + Thread::spawn(move|| { if i == 0 { tx.send(()); } else { f(i - 1, tx); } - }); + }).detach(); } f(10, tx); @@ -513,25 +505,25 @@ mod test { fn test_spawn_sched_childs_on_default_sched() { let (tx, rx) = channel(); - Thread::spawn(proc() { - Thread::spawn(proc() { + Thread::spawn(move|| { + Thread::spawn(move|| { tx.send(()); - }); - }); + }).detach(); + }).detach(); rx.recv(); } - fn avoid_copying_the_body(spawnfn: |v: proc():Send|) { + fn avoid_copying_the_body<F>(spawnfn: F) where F: FnOnce(Thunk) { let (tx, rx) = channel::<uint>(); let x = box 1; let x_in_parent = (&*x) as *const int as uint; - spawnfn(proc() { + spawnfn(Thunk::new(move|| { let x_in_child = (&*x) as *const int as uint; tx.send(x_in_child); - }); + })); let x_in_child = rx.recv(); assert_eq!(x_in_parent, x_in_child); @@ -539,24 +531,25 @@ mod test { #[test] fn test_avoid_copying_the_body_spawn() { - avoid_copying_the_body(|v| { Thread::spawn(v); }); + avoid_copying_the_body(|v| { + Thread::spawn(move || v.invoke(())).detach(); + }); } #[test] fn test_avoid_copying_the_body_thread_spawn() { avoid_copying_the_body(|f| { - let builder = cfg(); - builder.spawn(proc() { - f(); - }); + Thread::spawn(move|| { + f.invoke(()); + }).detach(); }) } #[test] fn test_avoid_copying_the_body_join() { avoid_copying_the_body(|f| { - let _ = Thread::with_join(proc() { - f() + let _ = Thread::spawn(move|| { + f.invoke(()) }).join(); }) } @@ -568,24 +561,24 @@ mod test { // (well, it would if the constant were 8000+ - I lowered it to be more // valgrind-friendly. try this at home, instead..!) static GENERATIONS: uint = 16; - fn child_no(x: uint) -> proc(): Send { - return proc() { + fn child_no(x: uint) -> Thunk { + return Thunk::new(move|| { if x < GENERATIONS { - Thread::spawn(child_no(x+1)); + Thread::spawn(move|| child_no(x+1).invoke(())).detach(); } - } + }); } - Thread::spawn(child_no(0)); + Thread::spawn(|| child_no(0).invoke(())).detach(); } #[test] fn test_simple_newsched_spawn() { - Thread::spawn(proc()()); + Thread::spawn(move || {}).detach(); } #[test] fn test_try_panic_message_static_str() { - match Thread::with_join(proc() { + match Thread::spawn(move|| { panic!("static string"); }).join() { Err(e) => { @@ -599,7 +592,7 @@ mod test { #[test] fn test_try_panic_message_owned_str() { - match Thread::with_join(proc() { + match Thread::spawn(move|| { panic!("owned string".to_string()); }).join() { Err(e) => { @@ -613,7 +606,7 @@ mod test { #[test] fn test_try_panic_message_any() { - match Thread::with_join(proc() { + match Thread::spawn(move|| { panic!(box 413u16 as Box<Any + Send>); }).join() { Err(e) => { @@ -631,7 +624,7 @@ mod test { fn test_try_panic_message_unit_struct() { struct Juju; - match Thread::with_join(proc() { + match Thread::spawn(move|| { panic!(Juju) }).join() { Err(ref e) if e.is::<Juju>() => {} @@ -645,9 +638,9 @@ mod test { let mut reader = ChanReader::new(rx); let stdout = ChanWriter::new(tx); - let r = cfg().stdout(box stdout as Box<Writer + Send>).with_join(proc() { - print!("Hello, world!"); - }).join(); + let r = Builder::new().stdout(box stdout as Box<Writer + Send>).spawn(move|| { + print!("Hello, world!"); + }).join(); assert!(r.is_ok()); let output = reader.read_to_string().unwrap(); diff --git a/src/libstd/thread_local/mod.rs b/src/libstd/thread_local/mod.rs index 4c052a4ce09..4c33d1c418d 100644 --- a/src/libstd/thread_local/mod.rs +++ b/src/libstd/thread_local/mod.rs @@ -68,6 +68,7 @@ pub mod scoped; /// /// ``` /// use std::cell::RefCell; +/// use std::thread::Thread; /// /// thread_local!(static FOO: RefCell<uint> = RefCell::new(1)); /// @@ -77,12 +78,12 @@ pub mod scoped; /// }); /// /// // each thread starts out with the initial value of 1 -/// spawn(move|| { +/// Thread::spawn(move|| { /// FOO.with(|f| { /// assert_eq!(*f.borrow(), 1); /// *f.borrow_mut() = 3; /// }); -/// }); +/// }).detach(); /// /// // we retain our original value of 2 despite the child thread /// FOO.with(|f| { @@ -533,7 +534,7 @@ mod tests { } } - Thread::with_join(move|| { + Thread::spawn(move|| { drop(S1); }).join(); } @@ -551,7 +552,7 @@ mod tests { } } - Thread::with_join(move|| unsafe { + Thread::spawn(move|| unsafe { K1.with(|s| *s.get() = Some(S1)); }).join(); } diff --git a/src/libstd/thunk.rs b/src/libstd/thunk.rs index 42e78495990..067926042f1 100644 --- a/src/libstd/thunk.rs +++ b/src/libstd/thunk.rs @@ -8,6 +8,9 @@ // option. This file may not be copied, modified, or distributed // except according to those terms. +// Because this module is temporary... +#![allow(missing_docs)] + use alloc::boxed::Box; use core::kinds::Send; use core::ops::FnOnce; |