src/libstd/task.rs

   1 // Copyright 2012-2013 The Rust Project Developers. See the COPYRIGHT
   2 // file at the top-level directory of this distribution and at
   3 // http://rust-lang.org/COPYRIGHT.
   4 //
   5 // Licensed under the Apache License, Version 2.0 <LICENSE-APACHE or
   6 // http://www.apache.org/licenses/LICENSE-2.0> or the MIT license
   7 // <LICENSE-MIT or http://opensource.org/licenses/MIT>, at your
   8 // option. This file may not be copied, modified, or distributed
   9 // except according to those terms.
  10
  11 //! Task creation
  12 //!
  13 //! An executing Rust program consists of a collection of tasks, each
  14 //! with their own stack and local state.
  15 //!
  16 //! Tasks generally have their memory *isolated* from each other by
  17 //! virtue of Rust's owned types (which of course may only be owned by
  18 //! a single task at a time). Communication between tasks is primarily
  19 //! done through [channels](../../std/comm/index.html), Rust's
  20 //! message-passing types, though [other forms of task
  21 //! synchronization](../../std/sync/index.html) are often employed to
  22 //! achieve particular performance goals. In particular, types that
  23 //! are guaranteed to be threadsafe are easily shared between threads
  24 //! using the atomically-reference-counted container,
  25 //! [`Arc`](../../std/sync/struct.Arc.html).
  26 //!
  27 //! Fatal logic errors in Rust cause *task panic*, during which
  28 //! a task will unwind the stack, running destructors and freeing
  29 //! owned resources. Task panic is unrecoverable from within
  30 //! the panicking task (i.e. there is no 'try/catch' in Rust), but
  31 //! panic may optionally be detected from a different task. If
  32 //! the main task panics the application will exit with a non-zero
  33 //! exit code.
  34 //!
  35 //! ## Example
  36 //!
  37 //! ```rust
  38 //! spawn(proc() {
  39 //!     println!("Hello, World!");
  40 //! })
  41 //! ```
  42
  43 #![unstable = "The task spawning model will be changed as part of runtime reform, and the module \
  44                will likely be renamed from `task` to `thread`."]
  45
  46 use any::Any;
  47 use borrow::IntoCow;
  48 use boxed::Box;
  49 use comm::channel;
  50 use io::{Writer, stdio};
  51 use kinds::{Send, marker};
  52 use option::Option;
  53 use option::Option::{None, Some};
  54 use result::Result;
  55 use rustrt::local::Local;
  56 use rustrt::task::Task;
  57 use rustrt::task;
  58 use str::{Str, SendStr};
  59 use string::{String, ToString};
  60 use sync::Future;
  61
  62 /// The task builder type.
  63 ///
  64 /// Provides detailed control over the properties and behavior of new tasks.
  65
  66 // NB: Builders are designed to be single-use because they do stateful
  67 // things that get weird when reusing - e.g. if you create a result future
  68 // it only applies to a single task, so then you have to maintain Some
  69 // potentially tricky state to ensure that everything behaves correctly
  70 // when you try to reuse the builder to spawn a new task. We'll just
  71 // sidestep that whole issue by making builders uncopyable and making
  72 // the run function move them in.
  73 pub struct TaskBuilder {
  74     // A name for the task-to-be, for identification in panic messages
  75     name: Option<SendStr>,
  76     // The size of the stack for the spawned task
  77     stack_size: Option<uint>,
  78     // Task-local stdout
  79     stdout: Option<Box<Writer + Send>>,
  80     // Task-local stderr
  81     stderr: Option<Box<Writer + Send>>,
  82     // Optionally wrap the eventual task body
  83     gen_body: Option<proc(v: proc():Send):Send -> proc():Send>,
  84     nocopy: marker::NoCopy,
  85 }
  86
  87 impl TaskBuilder {
  88     /// Generate the base configuration for spawning a task, off of which more
  89     /// configuration methods can be chained.
  90     pub fn new() -> TaskBuilder {
  91         TaskBuilder {
  92             name: None,
  93             stack_size: None,
  94             stdout: None,
  95             stderr: None,
  96             gen_body: None,
  97             nocopy: marker::NoCopy,
  98         }
  99     }
 100 }
 101
 102 impl TaskBuilder {
 103     /// Name the task-to-be. Currently the name is used for identification
 104     /// only in panic messages.
 105     #[unstable = "IntoMaybeOwned will probably change."]
 106     pub fn named<T: IntoCow<'static, String, str>>(mut self, name: T) -> TaskBuilder {
 107         self.name = Some(name.into_cow());
 108         self
 109     }
 110
 111     /// Set the size of the stack for the new task.
 112     pub fn stack_size(mut self, size: uint) -> TaskBuilder {
 113         self.stack_size = Some(size);
 114         self
 115     }
 116
 117     /// Redirect task-local stdout.
 118     #[experimental = "May not want to make stdio overridable here."]
 119     pub fn stdout(mut self, stdout: Box<Writer + Send>) -> TaskBuilder {
 120         self.stdout = Some(stdout);
 121         self
 122     }
 123
 124     /// Redirect task-local stderr.
 125     #[experimental = "May not want to make stdio overridable here."]
 126     pub fn stderr(mut self, stderr: Box<Writer + Send>) -> TaskBuilder {
 127         self.stderr = Some(stderr);
 128         self
 129     }
 130
 131     // Where spawning actually happens (whether yielding a future or not)
 132     fn spawn_internal(self, f: proc():Send,
 133                       on_exit: Option<proc(Result<(), Box<Any + Send>>):Send>) {
 134         let TaskBuilder {
 135             name, stack_size, stdout, stderr, mut gen_body, nocopy: _
 136         } = self;
 137         let f = match gen_body.take() {
 138             Some(gen) => gen(f),
 139             None => f
 140         };
 141         let opts = task::TaskOpts {
 142             on_exit: on_exit,
 143             name: name,
 144             stack_size: stack_size,
 145         };
 146         if stdout.is_some() || stderr.is_some() {
 147             Task::spawn(opts, proc() {
 148                 let _ = stdout.map(stdio::set_stdout);
 149                 let _ = stderr.map(stdio::set_stderr);
 150                 f();
 151             })
 152         } else {
 153             Task::spawn(opts, f)
 154         }
 155     }
 156
 157     /// Creates and executes a new child task.
 158     ///
 159     /// Sets up a new task with its own call stack and schedules it to run
 160     /// the provided proc. The task has the properties and behavior
 161     /// specified by the `TaskBuilder`.
 162     pub fn spawn(self, f: proc():Send) {
 163         self.spawn_internal(f, None)
 164     }
 165
 166     /// Execute a proc in a newly-spawned task and return a future representing
 167     /// the task's result. The task has the properties and behavior
 168     /// specified by the `TaskBuilder`.
 169     ///
 170     /// Taking the value of the future will block until the child task
 171     /// terminates.
 172     ///
 173     /// # Return value
 174     ///
 175     /// If the child task executes successfully (without panicking) then the
 176     /// future returns `result::Result::Ok` containing the value returned by the
 177     /// function. If the child task panics then the future returns
 178     /// `result::Result::Err` containing the argument to `panic!(...)` as an
 179     /// `Any` trait object.
 180     #[experimental = "Futures are experimental."]
 181     pub fn try_future<T:Send>(self, f: proc():Send -> T)
 182                               -> Future<Result<T, Box<Any + Send>>> {
 183         // currently, the on_exit proc provided by librustrt only works for unit
 184         // results, so we use an additional side-channel to communicate the
 185         // result.
 186
 187         let (tx_done, rx_done) = channel(); // signal that task has exited
 188         let (tx_retv, rx_retv) = channel(); // return value from task
 189
 190         let on_exit = proc(res) { let _ = tx_done.send_opt(res); };
 191         self.spawn_internal(proc() { let _ = tx_retv.send_opt(f()); },
 192                             Some(on_exit));
 193
 194         Future::from_fn(proc() {
 195             rx_done.recv().map(|_| rx_retv.recv())
 196         })
 197     }
 198
 199     /// Execute a function in a newly-spawnedtask and block until the task
 200     /// completes or panics. Equivalent to `.try_future(f).unwrap()`.
 201     #[unstable = "Error type may change."]
 202     pub fn try<T:Send>(self, f: proc():Send -> T) -> Result<T, Box<Any + Send>> {
 203         self.try_future(f).into_inner()
 204     }
 205 }
 206
 207 /* Convenience functions */
 208
 209 /// Creates and executes a new child task
 210 ///
 211 /// Sets up a new task with its own call stack and schedules it to run
 212 /// the provided unique closure.
 213 ///
 214 /// This function is equivalent to `TaskBuilder::new().spawn(f)`.
 215 pub fn spawn(f: proc(): Send) {
 216     TaskBuilder::new().spawn(f)
 217 }
 218
 219 /// Execute a function in a newly-spawned task and return either the return
 220 /// value of the function or an error if the task panicked.
 221 ///
 222 /// This is equivalent to `TaskBuilder::new().try`.
 223 #[unstable = "Error type may change."]
 224 pub fn try<T: Send>(f: proc(): Send -> T) -> Result<T, Box<Any + Send>> {
 225     TaskBuilder::new().try(f)
 226 }
 227
 228 /// Execute a function in another task and return a future representing the
 229 /// task's result.
 230 ///
 231 /// This is equivalent to `TaskBuilder::new().try_future`.
 232 #[experimental = "Futures are experimental."]
 233 pub fn try_future<T:Send>(f: proc():Send -> T) -> Future<Result<T, Box<Any + Send>>> {
 234     TaskBuilder::new().try_future(f)
 235 }
 236
 237
 238 /* Lifecycle functions */
 239
 240 /// Read the name of the current task.
 241 #[stable]
 242 pub fn name() -> Option<String> {
 243     use rustrt::task::Task;
 244
 245     let task = Local::borrow(None::<Task>);
 246     match task.name {
 247         Some(ref name) => Some(name.as_slice().to_string()),
 248         None => None
 249     }
 250 }
 251
 252 /// Yield control to the task scheduler.
 253 #[unstable = "Name will change."]
 254 pub fn deschedule() {
 255     use rustrt::task::Task;
 256     Task::yield_now();
 257 }
 258
 259 /// True if the running task is currently panicking (e.g. will return `true` inside a
 260 /// destructor that is run while unwinding the stack after a call to `panic!()`).
 261 #[unstable = "May move to a different module."]
 262 pub fn failing() -> bool {
 263     use rustrt::task::Task;
 264     Local::borrow(None::<Task>).unwinder.unwinding()
 265 }
 266
 267 #[cfg(test)]
 268 mod test {
 269     use any::{Any, AnyRefExt};
 270     use borrow::IntoCow;
 271     use boxed::BoxAny;
 272     use prelude::*;
 273     use result::Result::{Ok, Err};
 274     use result;
 275     use std::io::{ChanReader, ChanWriter};
 276     use string::String;
 277     use super::*;
 278
 279     // !!! These tests are dangerous. If something is buggy, they will hang, !!!
 280     // !!! instead of exiting cleanly. This might wedge the buildbots.       !!!
 281
 282     #[test]
 283     fn test_unnamed_task() {
 284         try(proc() {
 285             assert!(name().is_none());
 286         }).map_err(|_| ()).unwrap();
 287     }
 288
 289     #[test]
 290     fn test_owned_named_task() {
 291         TaskBuilder::new().named("ada lovelace".to_string()).try(proc() {
 292             assert!(name().unwrap() == "ada lovelace".to_string());
 293         }).map_err(|_| ()).unwrap();
 294     }
 295
 296     #[test]
 297     fn test_static_named_task() {
 298         TaskBuilder::new().named("ada lovelace").try(proc() {
 299             assert!(name().unwrap() == "ada lovelace".to_string());
 300         }).map_err(|_| ()).unwrap();
 301     }
 302
 303     #[test]
 304     fn test_send_named_task() {
 305         TaskBuilder::new().named("ada lovelace".into_cow()).try(proc() {
 306             assert!(name().unwrap() == "ada lovelace".to_string());
 307         }).map_err(|_| ()).unwrap();
 308     }
 309
 310     #[test]
 311     fn test_run_basic() {
 312         let (tx, rx) = channel();
 313         TaskBuilder::new().spawn(proc() {
 314             tx.send(());
 315         });
 316         rx.recv();
 317     }
 318
 319     #[test]
 320     fn test_try_future() {
 321         let result = TaskBuilder::new().try_future(proc() {});
 322         assert!(result.unwrap().is_ok());
 323
 324         let result = TaskBuilder::new().try_future(proc() -> () {
 325             panic!();
 326         });
 327         assert!(result.unwrap().is_err());
 328     }
 329
 330     #[test]
 331     fn test_try_success() {
 332         match try(proc() {
 333             "Success!".to_string()
 334         }).as_ref().map(|s| s.as_slice()) {
 335             result::Result::Ok("Success!") => (),
 336             _ => panic!()
 337         }
 338     }
 339
 340     #[test]
 341     fn test_try_panic() {
 342         match try(proc() {
 343             panic!()
 344         }) {
 345             result::Result::Err(_) => (),
 346             result::Result::Ok(()) => panic!()
 347         }
 348     }
 349
 350     #[test]
 351     fn test_spawn_sched() {
 352         use clone::Clone;
 353
 354         let (tx, rx) = channel();
 355
 356         fn f(i: int, tx: Sender<()>) {
 357             let tx = tx.clone();
 358             spawn(proc() {
 359                 if i == 0 {
 360                     tx.send(());
 361                 } else {
 362                     f(i - 1, tx);
 363                 }
 364             });
 365
 366         }
 367         f(10, tx);
 368         rx.recv();
 369     }
 370
 371     #[test]
 372     fn test_spawn_sched_childs_on_default_sched() {
 373         let (tx, rx) = channel();
 374
 375         spawn(proc() {
 376             spawn(proc() {
 377                 tx.send(());
 378             });
 379         });
 380
 381         rx.recv();
 382     }
 383
 384     fn avoid_copying_the_body(spawnfn: |v: proc():Send|) {
 385         let (tx, rx) = channel::<uint>();
 386
 387         let x = box 1;
 388         let x_in_parent = (&*x) as *const int as uint;
 389
 390         spawnfn(proc() {
 391             let x_in_child = (&*x) as *const int as uint;
 392             tx.send(x_in_child);
 393         });
 394
 395         let x_in_child = rx.recv();
 396         assert_eq!(x_in_parent, x_in_child);
 397     }
 398
 399     #[test]
 400     fn test_avoid_copying_the_body_spawn() {
 401         avoid_copying_the_body(spawn);
 402     }
 403
 404     #[test]
 405     fn test_avoid_copying_the_body_task_spawn() {
 406         avoid_copying_the_body(|f| {
 407             let builder = TaskBuilder::new();
 408             builder.spawn(proc() {
 409                 f();
 410             });
 411         })
 412     }
 413
 414     #[test]
 415     fn test_avoid_copying_the_body_try() {
 416         avoid_copying_the_body(|f| {
 417             let _ = try(proc() {
 418                 f()
 419             });
 420         })
 421     }
 422
 423     #[test]
 424     fn test_child_doesnt_ref_parent() {
 425         // If the child refcounts the parent task, this will stack overflow when
 426         // climbing the task tree to dereference each ancestor. (See #1789)
 427         // (well, it would if the constant were 8000+ - I lowered it to be more
 428         // valgrind-friendly. try this at home, instead..!)
 429         static GENERATIONS: uint = 16;
 430         fn child_no(x: uint) -> proc(): Send {
 431             return proc() {
 432                 if x < GENERATIONS {
 433                     TaskBuilder::new().spawn(child_no(x+1));
 434                 }
 435             }
 436         }
 437         TaskBuilder::new().spawn(child_no(0));
 438     }
 439
 440     #[test]
 441     fn test_simple_newsched_spawn() {
 442         spawn(proc()())
 443     }
 444
 445     #[test]
 446     fn test_try_panic_message_static_str() {
 447         match try(proc() {
 448             panic!("static string");
 449         }) {
 450             Err(e) => {
 451                 type T = &'static str;
 452                 assert!(e.is::<T>());
 453                 assert_eq!(*e.downcast::<T>().unwrap(), "static string");
 454             }
 455             Ok(()) => panic!()
 456         }
 457     }
 458
 459     #[test]
 460     fn test_try_panic_message_owned_str() {
 461         match try(proc() {
 462             panic!("owned string".to_string());
 463         }) {
 464             Err(e) => {
 465                 type T = String;
 466                 assert!(e.is::<T>());
 467                 assert_eq!(*e.downcast::<T>().unwrap(), "owned string".to_string());
 468             }
 469             Ok(()) => panic!()
 470         }
 471     }
 472
 473     #[test]
 474     fn test_try_panic_message_any() {
 475         match try(proc() {
 476             panic!(box 413u16 as Box<Any + Send>);
 477         }) {
 478             Err(e) => {
 479                 type T = Box<Any + Send>;
 480                 assert!(e.is::<T>());
 481                 let any = e.downcast::<T>().unwrap();
 482                 assert!(any.is::<u16>());
 483                 assert_eq!(*any.downcast::<u16>().unwrap(), 413u16);
 484             }
 485             Ok(()) => panic!()
 486         }
 487     }
 488
 489     #[test]
 490     fn test_try_panic_message_unit_struct() {
 491         struct Juju;
 492
 493         match try(proc() {
 494             panic!(Juju)
 495         }) {
 496             Err(ref e) if e.is::<Juju>() => {}
 497             Err(_) | Ok(()) => panic!()
 498         }
 499     }
 500
 501     #[test]
 502     fn test_stdout() {
 503         let (tx, rx) = channel();
 504         let mut reader = ChanReader::new(rx);
 505         let stdout = ChanWriter::new(tx);
 506
 507         let r = TaskBuilder::new().stdout(box stdout as Box<Writer + Send>)
 508                                   .try(proc() {
 509                 print!("Hello, world!");
 510             });
 511         assert!(r.is_ok());
 512
 513         let output = reader.read_to_string().unwrap();
 514         assert_eq!(output, "Hello, world!".to_string());
 515     }
 516
 517     // NOTE: the corresponding test for stderr is in run-pass/task-stderr, due
 518     // to the test harness apparently interfering with stderr configuration.
 519 }
 520
 521 #[test]
 522 fn task_abort_no_kill_runtime() {
 523     use std::io::timer;
 524     use time::Duration;
 525     use mem;
 526
 527     let tb = TaskBuilder::new();
 528     let rx = tb.try_future(proc() {});
 529     mem::drop(rx);
 530     timer::sleep(Duration::milliseconds(1000));
 531 }