Mirrors/rust
1
0
Fork 0
mirror of https://github.com/rust-lang/rust.git synced 2026-05-07 01:05:39 +03:00
Code Issues Packages Projects Releases Wiki Activity

Introduce std::thread

Browse Source 
Also removes:

* `std::task`
* `std::rt::task`
* `std::rt::thread`

Notes for the new API are in a follow-up commit.

Closes #18000
This commit is contained in:
Aaron Turon
2014-11-25 08:52:10 -08:00
parent 9b03b72d7f
commit cac133c9a8
6 changed files with 742 additions and 720 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/libstd/lib.rs
+2 -2
View File
@@ -227,9 +227,9 @@
pub mod collections;
pub mod hash;
/* Tasks and communication */
/* Threads and communication */
pub mod task;
pub mod thread;
pub mod sync;
pub mod comm;
src/libstd/rt/mod.rs
+15 -20
View File
@@ -53,7 +53,9 @@
use os;
use thunk::Thunk;
use kinds::Send;
use thread::Thread;
use sys_common;
use sys_common::thread::{mod, NewThread};
// Reexport some of our utilities which are expected by other crates.
pub use self::util::{default_sched_threads, min_stack, running_on_valgrind};
@@ -73,8 +75,6 @@
pub mod thread;
pub mod exclusive;
pub mod util;
pub mod local;
pub mod task;
pub mod unwind;
mod args;
@@ -98,8 +98,8 @@ pub fn init(argc: int, argv: *const *const u8) {
// Need to propagate the unsafety to `start`.
unsafe {
args::init(argc, argv);
local_ptr::init();
thread::init();
sys::thread::guard::init();
sys::stack_overflow::init();
unwind::register(failure::on_fail);
}
}
@@ -125,9 +125,6 @@ fn lang_start(main: *const u8, argc: int, argv: *const *const u8) -> int {
/// This procedure is guaranteed to run on the thread calling this function, but
/// the stack bounds for this rust task will *not* be set. Care must be taken
/// for this function to not overflow its stack.
///
/// This function will only return once *all* native threads in the system have
/// exited.
pub fn start(argc: int, argv: *const *const u8, main: Thunk) -> int {
use prelude::*;
use rt;
@@ -143,11 +140,9 @@ pub fn start(argc: int, argv: *const *const u8, main: Thunk) -> int {
// frames above our current position.
let my_stack_bottom = my_stack_top + 20000 - OS_DEFAULT_STACK_ESTIMATE;
// When using libgreen, one of the first things that we do is to turn off
// the SIGPIPE signal (set it to ignore). By default, some platforms will
// send a *signal* when a EPIPE error would otherwise be delivered. This
// runtime doesn't install a SIGPIPE handler, causing it to kill the
// program, which isn't exactly what we want!
// By default, some platforms will send a *signal* when a EPIPE error would
// otherwise be delivered. This runtime doesn't install a SIGPIPE handler,
// causing it to kill the program, which isn't exactly what we want!
//
// Hence, we set SIGPIPE to ignore when the program starts up in order to
// prevent this problem.
@@ -163,17 +158,18 @@ pub fn start(argc: int, argv: *const *const u8, main: Thunk) -> int {
init(argc, argv);
let mut exit_code = None;
let mut main = Some(main);
let mut task = box Task::new(Some((my_stack_bottom, my_stack_top)),
Some(rt::thread::main_guard_page()));
task.name = Some(str::Slice("<main>"));
drop(task.run(|| {
let thread: std::Thread = NewThread::new(Some("<main>".into_string()));
thread_info::set((my_stack_bottom, my_stack_top),
unsafe { sys::thread::guard::main() },
thread);
unwind::try(|| {
unsafe {
sys_common::stack::record_os_managed_stack_bounds(my_stack_bottom, my_stack_top);
}
(main.take().unwrap()).invoke(());
exit_code = Some(os::get_exit_status());
}).destroy());
});
unsafe { cleanup(); }
// If the exit code wasn't set, then the task block must have panicked.
return exit_code.unwrap_or(rt::DEFAULT_ERROR_CODE);
@@ -207,8 +203,7 @@ pub fn at_exit(f: proc():Send) {
/// undefined behavior.
pub unsafe fn cleanup() {
args::cleanup();
thread::cleanup();
local_ptr::cleanup();
sys::stack_overflow::cleanup();
}
// FIXME: these probably shouldn't be public...
src/libstd/rt/thread.rs
-171
View File
@@ -1,171 +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());
}
}
src/libstd/sys/common/thread_info.rs
+60
View File
@@ -0,0 +1,60 @@
// Copyright 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.
struct ThreadInfo {
// This field holds the known bounds of the stack in (lo, hi)
// form. Not all threads necessarily know their precise bounds,
// hence this is optional.
stack_bounds: (uint, uint),
stack_guard: uint,
unwinder: Unwinder,
thread: Thread,
}
thread_local!(static THREAD_INFO: RefCell<Option<ThreadInfo>> = RefCell::new(None));
impl ThreadInfo {
fn with<R>(f: |&ThreadInfo| -> R) -> R {
THREAD_INFO.with(|c| {
if c.borrow().is_none() {
*c.borrow_mut() = Some(ThreadInfo {
stack_bounds: (0, 0),
stack_guard: 0,
unwinder: Unwinder::new(),
thread: Thread::new(None),
})
}
f(c.borrow().as_ref().unwrap())
})
}
}
pub fn current_thread() -> Thread {
ThreadInfo::with(|info| info.thread.clone())
}
pub fn panicking() -> bool {
ThreadInfo::with(|info| info.unwinder.unwinding())
}
pub fn set(stack_bounds: (uint, uint), stack_guard: uint, thread: Thread) {
THREAD_INFO.with(|c| assert!(c.borrow().is_none()));
THREAD_INFO.with(|c| *c.borrow_mut() = Some(ThreadInfo{
stack_bounds: stack_bounds,
stack_guard: stack_guard,
unwinder: Unwinder::new(),
thread: thread,
}));
}
// a hack to get around privacy restrictions; implemented by `std::thread::Thread`
pub trait NewThread {
fn new(name: Option<String>) -> Self;
}
src/libstd/task.rs
+10 -527
View File
@@ -8,536 +8,19 @@
// option. This file may not be copied, modified, or distributed
// except according to those terms.
//! Task creation
//!
//! An executing Rust program consists of a collection of tasks, each
//! with their own stack and local state.
//!
//! Tasks 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 task at a time). Communication between tasks is primarily
//! done through [channels](../../std/comm/index.html), Rust's
//! message-passing types, though [other forms of task
//! synchronization](../../std/sync/index.html) are often employed to
//! achieve particular performance goals. 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).
//!
//! Fatal logic errors in Rust cause *task panic*, during which
//! a task will unwind the stack, running destructors and freeing
//! owned resources. Task panic is unrecoverable from within
//! the panicking task (i.e. there is no 'try/catch' in Rust), but
//! panic may optionally be detected from a different task. If
//! the main task panics the application will exit with a non-zero
//! exit code.
//!
//! # Examples
//!
//! ```rust
//! spawn(move|| {
//! println!("Hello, World!");
//! })
//! ```
//! Deprecated in favor of `thread`.
#![unstable = "The task spawning model will be changed as part of runtime reform, and the module \
will likely be renamed from `task` to `thread`."]
#![deprecated = "use std::thread instead"]
use any::Any;
use borrow::IntoCow;
use boxed::Box;
use comm::channel;
use core::ops::FnOnce;
use io::{Writer, stdio};
use thread;
use kinds::Send;
use option::Option;
use option::Option::{None, Some};
use result::Result;
use rt::local::Local;
use rt::task;
use rt::task::Task;
use str::SendStr;
use string::{String, ToString};
use thunk::{Thunk};
use sync::Future;
/// The task builder type.
///
/// Provides detailed control over the properties and behavior of new tasks.
/// Deprecate: use `std::thread::Cfg` instead.
#[deprecated = "use std::thread::Cfg instead"]
pub type TaskBuilder = thread::Cfg;
// NB: Builders are designed to be single-use because they do stateful
// things that get weird when reusing - e.g. if you create a result future
// it only applies to a single task, so then you have to maintain Some
// potentially tricky state to ensure that everything behaves correctly
// when you try to reuse the builder to spawn a new task. We'll just
// sidestep that whole issue by making builders uncopyable and making
// the run function move them in.
pub struct TaskBuilder {
// A name for the task-to-be, for identification in panic messages
name: Option<SendStr>,
// The size of the stack for the spawned task
stack_size: Option<uint>,
// Task-local stdout
stdout: Option<Box<Writer + Send>>,
// Task-local stderr
stderr: Option<Box<Writer + Send>>,
// Optionally wrap the eventual task body
gen_body: Option<Thunk<Thunk, Thunk>>,
}
impl TaskBuilder {
/// Generate the base configuration for spawning a task, off of which more
/// configuration methods can be chained.
pub fn new() -> TaskBuilder {
TaskBuilder {
name: None,
stack_size: None,
stdout: None,
stderr: None,
gen_body: None,
}
}
}
impl TaskBuilder {
/// Name the task-to-be. Currently the name is used for identification
/// only in panic messages.
#[unstable = "IntoMaybeOwned will probably change."]
pub fn named<T: IntoCow<'static, String, str>>(mut self, name: T) -> TaskBuilder {
self.name = Some(name.into_cow());
self
}
/// Set the size of the stack for the new task.
pub fn stack_size(mut self, size: uint) -> TaskBuilder {
self.stack_size = Some(size);
self
}
/// Redirect task-local stdout.
#[experimental = "May not want to make stdio overridable here."]
pub fn stdout(mut self, stdout: Box<Writer + Send>) -> TaskBuilder {
self.stdout = Some(stdout);
self
}
/// Redirect task-local stderr.
#[experimental = "May not want to make stdio overridable here."]
pub fn stderr(mut self, stderr: Box<Writer + Send>) -> TaskBuilder {
self.stderr = Some(stderr);
self
}
// Where spawning actually happens (whether yielding a future or not)
fn spawn_internal(
self,
f: Thunk,
on_exit: Option<Thunk<task::Result>>)
{
let TaskBuilder {
name, stack_size, stdout, stderr, mut gen_body
} = self;
let f = match gen_body.take() {
Some(gen) => gen.invoke(f),
None => f
};
let opts = task::TaskOpts {
on_exit: on_exit,
name: name,
stack_size: stack_size,
};
if stdout.is_some() || stderr.is_some() {
Task::spawn(opts, move|:| {
let _ = stdout.map(stdio::set_stdout);
let _ = stderr.map(stdio::set_stderr);
f.invoke(());
});
} else {
Task::spawn(opts, move|:| f.invoke(()))
}
}
/// Creates and executes a new child task.
///
/// Sets up a new task with its own call stack and schedules it to run
/// the provided function. The task has the properties and behavior
/// specified by the `TaskBuilder`.
pub fn spawn<F:FnOnce()+Send>(self, f: F) {
self.spawn_internal(Thunk::new(f), None)
}
/// Execute a function in a newly-spawned task and return a future representing
/// the task's result. The task has the properties and behavior
/// specified by the `TaskBuilder`.
///
/// Taking the value of the future will block until the child task
/// terminates.
///
/// # Return value
///
/// If the child task executes successfully (without panicking) then the
/// future returns `result::Result::Ok` containing the value returned by the
/// function. If the child task panics then the future returns
/// `result::Result::Err` containing the argument to `panic!(...)` as an
/// `Any` trait object.
#[experimental = "Futures are experimental."]
pub fn try_future<T:Send,F:FnOnce()->(T)+Send>(self, f: F)
-> Future<Result<T, Box<Any + Send>>> {
// currently, the on_exit fn provided by librustrt only works for unit
// results, so we use an additional side-channel to communicate the
// result.
let (tx_done, rx_done) = channel(); // signal that task has exited
let (tx_retv, rx_retv) = channel(); // return value from task
let on_exit: Thunk<task::Result> = Thunk::with_arg(move |: res: task::Result| {
let _ = tx_done.send_opt(res);
});
self.spawn_internal(Thunk::new(move |:| { let _ = tx_retv.send_opt(f()); }),
Some(on_exit));
Future::from_fn(move|:| {
rx_done.recv().map(|_| rx_retv.recv())
})
}
/// Execute a function in a newly-spawnedtask and block until the task
/// completes or panics. Equivalent to `.try_future(f).unwrap()`.
#[unstable = "Error type may change."]
pub fn try<T,F>(self, f: F) -> Result<T, Box<Any + Send>>
where F : FnOnce() -> T, F : Send, T : Send
{
self.try_future(f).into_inner()
}
}
/* Convenience functions */
/// Creates and executes a new child task
///
/// Sets up a new task with its own call stack and schedules it to run
/// the provided unique closure.
///
/// This function is equivalent to `TaskBuilder::new().spawn(f)`.
pub fn spawn<F:FnOnce()+Send>(f: F) {
TaskBuilder::new().spawn(f)
}
/// Execute a function in a newly-spawned task and return either the return
/// value of the function or an error if the task panicked.
///
/// This is equivalent to `TaskBuilder::new().try`.
#[unstable = "Error type may change."]
pub fn try<T,F>(f: F) -> Result<T, Box<Any + Send>>
where T : Send, F : FnOnce() -> T, F : Send
{
TaskBuilder::new().try(f)
}
/// Execute a function in another task and return a future representing the
/// task's result.
///
/// This is equivalent to `TaskBuilder::new().try_future`.
#[experimental = "Futures are experimental."]
pub fn try_future<T,F>(f: F) -> Future<Result<T, Box<Any + Send>>>
where T:Send, F:FnOnce()->T, F:Send
{
TaskBuilder::new().try_future(f)
}
/* Lifecycle functions */
/// Read the name of the current task.
#[stable]
pub fn name() -> Option<String> {
use rt::task::Task;
let task = Local::borrow(None::<Task>);
match task.name {
Some(ref name) => Some(name.to_string()),
None => None
}
}
/// Yield control to the task scheduler.
#[unstable = "Name will change."]
pub fn deschedule() {
use rt::task::Task;
Task::yield_now();
}
/// True if the running task is currently panicking (e.g. will return `true` inside a
/// destructor that is run while unwinding the stack after a call to `panic!()`).
#[unstable = "May move to a different module."]
pub fn failing() -> bool {
use rt::task::Task;
Local::borrow(None::<Task>).unwinder.unwinding()
}
#[cfg(test)]
mod test {
use any::{Any, AnyRefExt};
use borrow::IntoCow;
use boxed::BoxAny;
use prelude::*;
use result::Result::{Ok, Err};
use result;
use std::io::{ChanReader, ChanWriter};
use string::String;
use thunk::Thunk;
use prelude::*;
use super::*;
// !!! These tests are dangerous. If something is buggy, they will hang, !!!
// !!! instead of exiting cleanly. This might wedge the buildbots. !!!
#[test]
fn test_unnamed_task() {
try(move|| {
assert!(name().is_none());
}).map_err(|_| ()).unwrap();
}
#[test]
fn test_owned_named_task() {
TaskBuilder::new().named("ada lovelace".to_string()).try(move|| {
assert!(name().unwrap() == "ada lovelace");
}).map_err(|_| ()).unwrap();
}
#[test]
fn test_static_named_task() {
TaskBuilder::new().named("ada lovelace").try(move|| {
assert!(name().unwrap() == "ada lovelace");
}).map_err(|_| ()).unwrap();
}
#[test]
fn test_send_named_task() {
TaskBuilder::new().named("ada lovelace".into_cow()).try(move|| {
assert!(name().unwrap() == "ada lovelace");
}).map_err(|_| ()).unwrap();
}
#[test]
fn test_run_basic() {
let (tx, rx) = channel();
TaskBuilder::new().spawn(move|| {
tx.send(());
});
rx.recv();
}
#[test]
fn test_try_future() {
let result = TaskBuilder::new().try_future(move|| {});
assert!(result.unwrap().is_ok());
let result = TaskBuilder::new().try_future(move|| -> () {
panic!();
});
assert!(result.unwrap().is_err());
}
#[test]
fn test_try_success() {
match try(move|| {
"Success!".to_string()
}).as_ref().map(|s| s.as_slice()) {
result::Result::Ok("Success!") => (),
_ => panic!()
}
}
#[test]
fn test_try_panic() {
match try(move|| {
panic!()
}) {
result::Result::Err(_) => (),
result::Result::Ok(()) => panic!()
}
}
#[test]
fn test_spawn_sched() {
use clone::Clone;
let (tx, rx) = channel();
fn f(i: int, tx: Sender<()>) {
let tx = tx.clone();
spawn(move|| {
if i == 0 {
tx.send(());
} else {
f(i - 1, tx);
}
});
}
f(10, tx);
rx.recv();
}
#[test]
fn test_spawn_sched_childs_on_default_sched() {
let (tx, rx) = channel();
spawn(move|| {
spawn(move|| {
tx.send(());
});
});
rx.recv();
}
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(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);
}
#[test]
fn test_avoid_copying_the_body_spawn() {
avoid_copying_the_body(|t| spawn(move|| t.invoke(())));
}
#[test]
fn test_avoid_copying_the_body_task_spawn() {
avoid_copying_the_body(|f| {
let builder = TaskBuilder::new();
builder.spawn(move|| f.invoke(()));
})
}
#[test]
fn test_avoid_copying_the_body_try() {
avoid_copying_the_body(|f| {
let _ = try(move|| f.invoke(()));
})
}
#[test]
fn test_child_doesnt_ref_parent() {
// If the child refcounts the parent task, this will stack overflow when
// climbing the task tree to dereference each ancestor. (See #1789)
// (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) -> Thunk {
return Thunk::new(move|| {
if x < GENERATIONS {
TaskBuilder::new().spawn(move|| child_no(x+1).invoke(()));
}
});
}
TaskBuilder::new().spawn(|| child_no(0).invoke(()));
}
#[test]
fn test_simple_newsched_spawn() {
spawn(move|| ())
}
#[test]
fn test_try_panic_message_static_str() {
match try(move|| {
panic!("static string");
}) {
Err(e) => {
type T = &'static str;
assert!(e.is::<T>());
assert_eq!(*e.downcast::<T>().unwrap(), "static string");
}
Ok(()) => panic!()
}
}
#[test]
fn test_try_panic_message_owned_str() {
match try(move|| {
panic!("owned string".to_string());
}) {
Err(e) => {
type T = String;
assert!(e.is::<T>());
assert_eq!(*e.downcast::<T>().unwrap(), "owned string");
}
Ok(()) => panic!()
}
}
#[test]
fn test_try_panic_message_any() {
match try(move|| {
panic!(box 413u16 as Box<Any + Send>);
}) {
Err(e) => {
type T = Box<Any + Send>;
assert!(e.is::<T>());
let any = e.downcast::<T>().unwrap();
assert!(any.is::<u16>());
assert_eq!(*any.downcast::<u16>().unwrap(), 413u16);
}
Ok(()) => panic!()
}
}
#[test]
fn test_try_panic_message_unit_struct() {
struct Juju;
match try(move|| {
panic!(Juju)
}) {
Err(ref e) if e.is::<Juju>() => {}
Err(_) | Ok(()) => panic!()
}
}
#[test]
fn test_stdout() {
let (tx, rx) = channel();
let mut reader = ChanReader::new(rx);
let stdout = ChanWriter::new(tx);
let r = TaskBuilder::new().stdout(box stdout as Box<Writer + Send>)
.try(move|| {
print!("Hello, world!");
});
assert!(r.is_ok());
let output = reader.read_to_string().unwrap();
assert_eq!(output, "Hello, world!");
}
// NOTE: the corresponding test for stderr is in run-pass/task-stderr, due
// to the test harness apparently interfering with stderr configuration.
}
#[test]
fn task_abort_no_kill_runtime() {
use std::io::timer;
use time::Duration;
use mem;
let tb = TaskBuilder::new();
let rx = tb.try_future(move|| {});
mem::drop(rx);
timer::sleep(Duration::milliseconds(1000));
/// Deprecated: use `std::thread::Thread::spawn` instead.
#[deprecated = "use std::thread::Thread::spawn instead"]
pub fn spawn(f: proc(): Send) {
thread::Thread::spawn(f);
}
src/libstd/thread.rs
+655
View File
@@ -0,0 +1,655 @@
// Copyright 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 threads
//!
//! ## The threading model
//!
//! 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
//! atomically-reference-counted container,
//! [`Arc`](../../std/sync/struct.Arc.html).
//!
//! Fatal logic errors in Rust cause *thread panic*, during which
//! a thread will unwind the stack, running destructors and freeing
//! owned resources. Thread panic is unrecoverable from within
//! the panicking thread (i.e. there is no 'try/catch' in Rust), but
//! panic may optionally be detected from a different thread. If
//! the main thread panics the application will exit with a non-zero
//! exit code.
//!
//! When the main thread of a Rust program terminates, the entire program shuts
//! down, even if other threads are still running. However, this module provides
//! convenient facilities for automatically waiting for the termination of a
//! child thread (i.e., join), described below.
//!
//! ## The `Thread` type
//!
//! Already-running threads are represented via the `Thread` type, which you can
//! get in one of two ways:
//!
//! * By spawning a new thread, e.g. using the `Thread::spawn` constructor;
//! * By requesting the current thread, using the `Thread::current` function.
//!
//! Threads can be named, and provide some built-in support for low-level
//! synchronization described below.
//!
//! The `Thread::current()` function is available even for threads not spawned
//! by the APIs of this module.
//!
//! ## 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:
//!
//! ```rust
//! use std::thread::Thread;
//!
//! Thread::spawn(proc() {
//! println!("Hello, World!");
//! })
//! ```
//!
//! 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;
//!
//! let guard = Thread::with_join(proc() { panic!() };
//! assert!(guard.join().is_err());
//! ```
//!
//! 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.
//!
//! ### Configured 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:
//!
//! ```rust
//! use std::thread;
//!
//! thread::cfg().name("child1").spawn(proc() { println!("Hello, world!") });
//! ```
//!
//! ## Blocking support: park and unpark
//!
//! Every thread is equipped with some basic low-level blocking support, via the
//! `park` and `unpark` functions.
//!
//! Conceptually, each `Thread` handle has an associated token, which is
//! initially not present:
//!
//! * The `Thread::park()` function blocks the current thread unless or until
//! the token is available for its thread handle, at which point It atomically
//! consumes the token. It may also return *spuriously*, without consuming the
//! token.
//!
//! * The `unpark()` method on a `Thread` atomically makes the token available
//! if it wasn't already.
//!
//! In other words, each `Thread` acts a bit like a semaphore with initial count
//! 0, except that the semaphore is *saturating* (the count cannot go above 1),
//! and can return spuriously.
//!
//! The API is typically used by acquiring a handle to the current thread,
//! placing that handle in a shared data structure so that other threads can
//! find it, and then `park`ing. When some desired condition is met, another
//! thread calls `unpark` on the handle.
//!
//! The motivation for this design is twofold:
//!
//! * It avoids the need to allocate mutexes and condvars when building new
//! synchronization primitives; the threads already provide basic blocking/signaling.
//!
//! * It can be implemented highly efficiently on many platforms.
use core::prelude::*;
use any::Any;
use borrow::IntoCow;
use boxed::Box;
use mem;
use sync::{Mutex, Condvar, Arc};
use string::String;
use rt::{mod, unwind};
use io::{Writer, stdio};
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 {
// A name for the thread-to-be, for identification in panic messages
name: Option<String>,
// The size of the stack for the spawned thread
stack_size: Option<uint>,
// Thread-local stdout
stdout: Option<Box<Writer + Send>>,
// Thread-local stderr
stderr: Option<Box<Writer + Send>>,
}
impl Cfg {
/// Generate the base configuration for spawning a thread, from which
/// configuration methods can be chained.
pub fn new() -> Cfg {
Cfg {
name: None,
stack_size: None,
stdout: None,
stderr: None,
}
}
/// Name the thread-to-be. Currently the name is used for identification
/// only in panic messages.
pub fn name(mut self, name: String) -> Cfg {
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 {
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 {
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 {
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 {
self.stderr = Some(stderr);
self
}
fn core_spawn<T: Send>(self, f: proc():Send -> T, after: proc(Result<T>):Send)
-> (imp::rust_thread, Thread)
{
let Cfg { name, stack_size, stdout, stderr } = self;
let stack_size = stack_size.unwrap_or(rt::min_stack());
let my_thread = Thread::new(name);
let their_thread = my_thread.clone();
// Spawning a new OS thread guarantees that __morestack will never get
// triggered, but we must manually set up the actual stack bounds once
// this function starts executing. This raises the lower limit by a bit
// 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 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;
let my_stack_bottom = my_stack_top - stack_size + 1024;
unsafe {
stack::record_os_managed_stack_bounds(my_stack_bottom, my_stack_top);
}
thread_info::set(
(my_stack_bottom, my_stack_top),
thread::current_guard_page(),
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.
unsafe {
let mut output = None;
let mut f_opt = Some( // option dance
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 = unwind::try(|| output = Some((f_opt.take().unwrap())()));
match (output, try_result) {
(Some(data), Ok(_)) => after(Ok(data)),
(None, Err(cause)) => after(Err(cause)),
_ => unreachable!()
}
}
};
(unsafe { imp::create(stack, 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 my_packet = box Err(box 0); // sentinel value
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) {
*their_packet = result;
});
JoinGuard {
native: native,
joined: false,
packet: my_packet,
thread: 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
cvar: Condvar,
}
#[deriving(Clone)]
/// A handle to a thread.
pub struct Thread {
inner: Arc<Inner>,
}
impl Thread {
fn new(name: Option<String>) -> Thread {
Thread {
inner: Arc::new(Inner {
name: name,
lock: Mutex::new(false),
cvar: Condvar::new(),
})
}
}
/// Spawn a detached thread, and return a handle to 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)
}
/// Gets a handle to the thread that invokes it.
pub fn current() -> Thread {
ThreadInfo::current_thread()
}
/// Cooperatively give up a timeslice to the OS scheduler.
pub fn yield_now() {
unsafe { imp::yield_now() }
}
/// Determines whether the current thread is panicking.
pub fn panicking() -> bool {
ThreadInfo::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.
pub fn park() {
let thread = Thread::current();
let guard = thread.inner.lock.lock();
while !*guard {
thread.inner.cvar.wait(guard);
}
*guard = false;
}
/// Atomically makes the handle's token available if it is not already.
///
/// See the module doc for more detail.
pub fn unpark(&self) {
let guard = self.inner.lock();
if !*guard {
*guard = true;
self.inner.cvar.notify_one();
}
}
/// Get the thread's name.
pub fn name(&self) -> Option<&str> {
self.inner.name.as_ref()
}
}
// a hack to get around privacy restrictions
impl thread_info::NewThread for Thread {
fn new(name: Option<String>) -> Thread { Thread::new(name) }
}
/// Indicates the manner in which a thread exited.
///
/// A thread that completes without panicking is considered to exit successfully.
pub type Result<T> = result::Result<T, Box<Any + Send>>;
#[must_use]
/// An RAII guard that will block until thread termination when dropped.
pub struct JoinGuard<T> {
native: imp::rust_thread,
thread: Thread,
joined: bool,
packet: Box<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()
}
/// Wait for the associated thread to finish, returning the result of the thread's
/// calculation.
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_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};
use boxed::BoxAny;
use prelude::*;
use result::Result::{Ok, Err};
use result;
use std::io::{ChanReader, ChanWriter};
use string::String;
use super::{Thread, cfg};
// !!! 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() {
assert!(Thread::current().name().is_none());
}).join().map_err(|_| ()).unwrap();
}
#[test]
fn test_named_thread() {
cfg().name("ada lovelace".to_string()).with_join(proc() {
assert!(Thread::current().name().unwrap() == "ada lovelace".to_string());
}).join().map_err(|_| ()).unwrap();
}
#[test]
fn test_run_basic() {
let (tx, rx) = channel();
Thread::spawn(proc() {
tx.send(());
});
rx.recv();
}
#[test]
fn test_join_success() {
match Thread::with_join::<String>(proc() {
"Success!".to_string()
}).join().as_ref().map(|s| s.as_slice()) {
result::Result::Ok("Success!") => (),
_ => panic!()
}
}
#[test]
fn test_join_panic() {
match Thread::with_join(proc() {
panic!()
}).join() {
result::Result::Err(_) => (),
result::Result::Ok(()) => panic!()
}
}
#[test]
fn test_spawn_sched() {
use clone::Clone;
let (tx, rx) = channel();
fn f(i: int, tx: Sender<()>) {
let tx = tx.clone();
Thread::spawn(proc() {
if i == 0 {
tx.send(());
} else {
f(i - 1, tx);
}
});
}
f(10, tx);
rx.recv();
}
#[test]
fn test_spawn_sched_childs_on_default_sched() {
let (tx, rx) = channel();
Thread::spawn(proc() {
Thread::spawn(proc() {
tx.send(());
});
});
rx.recv();
}
fn avoid_copying_the_body(spawnfn: |v: proc():Send|) {
let (tx, rx) = channel::<uint>();
let x = box 1;
let x_in_parent = (&*x) as *const int as uint;
spawnfn(proc() {
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);
}
#[test]
fn test_avoid_copying_the_body_spawn() {
avoid_copying_the_body(|v| { Thread::spawn(v); });
}
#[test]
fn test_avoid_copying_the_body_thread_spawn() {
avoid_copying_the_body(|f| {
let builder = cfg();
builder.spawn(proc() {
f();
});
})
}
#[test]
fn test_avoid_copying_the_body_join() {
avoid_copying_the_body(|f| {
let _ = Thread::with_join(proc() {
f()
}).join();
})
}
#[test]
fn test_child_doesnt_ref_parent() {
// If the child refcounts the parent task, this will stack overflow when
// climbing the task tree to dereference each ancestor. (See #1789)
// (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() {
if x < GENERATIONS {
Thread::spawn(child_no(x+1));
}
}
}
Thread::spawn(child_no(0));
}
#[test]
fn test_simple_newsched_spawn() {
Thread::spawn(proc()());
}
#[test]
fn test_try_panic_message_static_str() {
match Thread::with_join(proc() {
panic!("static string");
}).join() {
Err(e) => {
type T = &'static str;
assert!(e.is::<T>());
assert_eq!(*e.downcast::<T>().unwrap(), "static string");
}
Ok(()) => panic!()
}
}
#[test]
fn test_try_panic_message_owned_str() {
match Thread::with_join(proc() {
panic!("owned string".to_string());
}).join() {
Err(e) => {
type T = String;
assert!(e.is::<T>());
assert_eq!(*e.downcast::<T>().unwrap(), "owned string".to_string());
}
Ok(()) => panic!()
}
}
#[test]
fn test_try_panic_message_any() {
match Thread::with_join(proc() {
panic!(box 413u16 as Box<Any + Send>);
}).join() {
Err(e) => {
type T = Box<Any + Send>;
assert!(e.is::<T>());
let any = e.downcast::<T>().unwrap();
assert!(any.is::<u16>());
assert_eq!(*any.downcast::<u16>().unwrap(), 413u16);
}
Ok(()) => panic!()
}
}
#[test]
fn test_try_panic_message_unit_struct() {
struct Juju;
match Thread::with_join(proc() {
panic!(Juju)
}).join() {
Err(ref e) if e.is::<Juju>() => {}
Err(_) | Ok(()) => panic!()
}
}
#[test]
fn test_stdout() {
let (tx, rx) = channel();
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();
assert!(r.is_ok());
let output = reader.read_to_string().unwrap();
assert_eq!(output, "Hello, world!".to_string());
}
// NOTE: the corresponding test for stderr is in run-pass/task-stderr, due
// to the test harness apparently interfering with stderr configuration.
}